Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
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
3 changes: 2 additions & 1 deletion gradle/instrumentation.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,8 @@ afterEvaluate {
jvmArgs "-javaagent:${project(":testing:agent-for-testing").tasks.shadowJar.archiveFile.get().asFile.absolutePath}"
jvmArgs "-Dotel.javaagent.experimental.initializer.jar=${shadowJar.archiveFile.get().asFile.absolutePath}"
jvmArgs "-Dotel.javaagent.testing.additional-library-ignores.enabled=false"
jvmArgs "-Dotel.javaagent.testing.fail-on-context-leak=true"
def failOnContextLeak = findProperty('failOnContextLeak')
jvmArgs "-Dotel.javaagent.testing.fail-on-context-leak=${failOnContextLeak == null || failOnContextLeak}"
// prevent sporadic gradle deadlocks, see SafeLogger for more details
jvmArgs "-Dotel.javaagent.testing.transform-safe-logging.enabled=true"

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import io.opentelemetry.instrumentation.api.InstrumentationVersion;
import io.opentelemetry.instrumentation.api.internal.SupportabilityMetrics;
import io.opentelemetry.instrumentation.api.tracer.ClientSpan;
import io.opentelemetry.instrumentation.api.tracer.ConsumerSpan;
import io.opentelemetry.instrumentation.api.tracer.ServerSpan;
import java.util.ArrayList;
import java.util.List;
Expand Down Expand Up @@ -89,10 +90,11 @@ public boolean shouldStart(Context parentContext, REQUEST request) {
SpanKind spanKind = spanKindExtractor.extract(request);
switch (spanKind) {
case SERVER:
suppressed = ServerSpan.fromContextOrNull(parentContext) != null;
case CONSUMER:
suppressed = ServerSpan.exists(parentContext) || ConsumerSpan.exists(parentContext);
Comment on lines 92 to +94

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

SERVER and CONSUMER spans are now both treated as "local-root" type of spans; by default, you can't have both of them in a single process, in a single trace, since they both represent the receiving side.
This is a pretty significant change, please tell me WDYT.

break;
case CLIENT:
suppressed = ClientSpan.fromContextOrNull(parentContext) != null;
suppressed = ClientSpan.exists(parentContext);
break;
default:
break;
Expand Down Expand Up @@ -146,6 +148,8 @@ public Context start(Context parentContext, REQUEST request) {
return ServerSpan.with(context, span);
case CLIENT:
return ClientSpan.with(context, span);
case CONSUMER:
return ConsumerSpan.with(context, span);
default:
return context;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -104,10 +104,11 @@ public final boolean shouldStartSpan(Context context, SpanKind proposedKind) {
boolean suppressed = false;
switch (proposedKind) {
case CLIENT:
suppressed = inClientSpan(context);
suppressed = ClientSpan.exists(context);
break;
case SERVER:
suppressed = inServerSpan(context);
case CONSUMER:
suppressed = ServerSpan.exists(context) || ConsumerSpan.exists(context);
break;
default:
break;
Expand All @@ -118,14 +119,6 @@ public final boolean shouldStartSpan(Context context, SpanKind proposedKind) {
return !suppressed;
}

private static boolean inClientSpan(Context context) {
return ClientSpan.fromContextOrNull(context) != null;
}

private static boolean inServerSpan(Context context) {
return ServerSpan.fromContextOrNull(context) != null;
}

/**
* Returns a {@link Context} inheriting from {@code Context.current()} that contains a new span
* with name {@code spanName} and kind {@link SpanKind#INTERNAL}.
Expand Down Expand Up @@ -176,6 +169,16 @@ protected final Context withServerSpan(Context parentContext, Span span) {
return ServerSpan.with(parentContext.with(span), span);
}

/**
* Returns a {@link Context} containing the passed {@code span} marked as the current {@link
* SpanKind#CONSUMER} span.
*
* @see #shouldStartSpan(Context, SpanKind)
*/
protected final Context withConsumerSpan(Context parentContext, Span span) {
return ConsumerSpan.with(parentContext.with(span), span);
}

/** Ends the execution of a span stored in the passed {@code context}. */
public void end(Context context) {
end(context, -1);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,20 @@
import io.opentelemetry.context.ContextKey;
import org.checkerframework.checker.nullness.qual.Nullable;

/**
* This class encapsulates the context key for storing the current {@link SpanKind#CLIENT} span in
* the {@link Context}.
*/
public final class ClientSpan {
// Keeps track of the client span in a subtree corresponding to a client request.
private static final ContextKey<Span> KEY =
ContextKey.named("opentelemetry-traces-client-span-key");

/** Returns true when a {@link SpanKind#CLIENT} span is present in the passed {@code context}. */
public static boolean exists(Context context) {
return fromContextOrNull(context) != null;
}

/**
* Returns span of type {@link SpanKind#CLIENT} from the given context or {@code null} if not
* found.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/*
* Copyright The OpenTelemetry Authors
* SPDX-License-Identifier: Apache-2.0
*/

package io.opentelemetry.instrumentation.api.tracer;

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.SpanKind;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.ContextKey;
import org.checkerframework.checker.nullness.qual.Nullable;

/**
* This class encapsulates the context key for storing the current {@link SpanKind#CONSUMER} span in
* the {@link Context}.
*/
public final class ConsumerSpan {
// Keeps track of the consumer span for the current trace.
private static final ContextKey<Span> KEY =
ContextKey.named("opentelemetry-traces-consumer-span-key");

/**
* Returns true when a {@link SpanKind#CONSUMER} span is present in the passed {@code context}.
*/
public static boolean exists(Context context) {
return fromContextOrNull(context) != null;
}

/**
* Returns span of type {@link SpanKind#CONSUMER} from the given context or {@code null} if not
* found.
*/
@Nullable
public static Span fromContextOrNull(Context context) {
return context.get(KEY);
}

public static Context with(Context context, Span consumerSpan) {
return context.with(KEY, consumerSpan);
}

private ConsumerSpan() {}
}
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,11 @@ public final class ServerSpan {
private static final ContextKey<Span> KEY =
ContextKey.named("opentelemetry-traces-server-span-key");

/** Returns true when a {@link SpanKind#SERVER} span is present in the passed {@code context}. */
public static boolean exists(Context context) {
return fromContextOrNull(context) != null;
}

/**
* Returns span of type {@link SpanKind#SERVER} from the given context or {@code null} if not
* found.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ class BaseTracerTest extends Specification {

where:
kind | context | expected
SpanKind.CLIENT | root | true
SpanKind.CLIENT | root | true
SpanKind.SERVER | root | true
SpanKind.INTERNAL | root | true
SpanKind.PRODUCER | root | true
Expand All @@ -54,7 +54,7 @@ class BaseTracerTest extends Specification {
SpanKind.PRODUCER | tracer.withClientSpan(root, existingSpan) | true
SpanKind.SERVER | tracer.withServerSpan(root, existingSpan) | false
SpanKind.INTERNAL | tracer.withServerSpan(root, existingSpan) | true
SpanKind.CONSUMER | tracer.withServerSpan(root, existingSpan) | true
SpanKind.CONSUMER | tracer.withServerSpan(root, existingSpan) | false
SpanKind.PRODUCER | tracer.withServerSpan(root, existingSpan) | true
SpanKind.CLIENT | tracer.withServerSpan(root, existingSpan) | true
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public Context startSpan(ConsumerRecord<?, ?> record) {
.startSpan();

onConsume(span, now, record);
return parentContext.with(span);
return withConsumerSpan(parentContext, span);
}

private Context extractParent(ConsumerRecord<?, ?> record) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ public Context startSpan(StampedRecord record) {
.setAttribute(SemanticAttributes.MESSAGING_OPERATION, "process")
.startSpan();
onConsume(span, record);
return parentContext.with(span);
return withConsumerSpan(parentContext, span);
}

public String spanNameForConsume(StampedRecord record) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -145,10 +145,7 @@ public static void setSpanNameAddHeaders(
if (props == null) {
props = MessageProperties.MINIMAL_BASIC;
}
Integer deliveryMode = props.getDeliveryMode();
if (deliveryMode != null) {
span.setAttribute("rabbitmq.delivery_mode", deliveryMode);
}
tracer().onProps(span, props);

// We need to copy the BasicProperties and provide a header map we can modify
Map<String, Object> headers = props.getHeaders();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ public Context startDeliverySpan(
"rabbitmq.record.queue_time_ms", Math.max(0L, startTimeMillis - produceTimeMillis));
}

return parentContext.with(span);
return withConsumerSpan(parentContext, span);
}

public void onPublish(Span span, String exchange, String routingKey) {
Expand All @@ -126,6 +126,15 @@ public void onPublish(Span span, String exchange, String routingKey) {
}
}

public void onProps(Span span, AMQP.BasicProperties props) {
if (CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES) {
Integer deliveryMode = props.getDeliveryMode();
if (deliveryMode != null) {
span.setAttribute("rabbitmq.delivery_mode", deliveryMode);
}
}
}

public String spanNameOnGet(String queue) {
return (queue.startsWith("amq.gen-") ? "<generated>" : queue) + " receive";
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,13 +40,13 @@ protected String getInstrumentationName() {
Context startSpan(Context parentContext, List<MessageExt> msgs) {
if (msgs.size() == 1) {
SpanBuilder spanBuilder = startSpanBuilder(extractParent(msgs.get(0)), msgs.get(0));
return parentContext.with(spanBuilder.startSpan());
return withConsumerSpan(parentContext, spanBuilder.startSpan());
} else {
SpanBuilder spanBuilder =
spanBuilder(parentContext, "multiple_sources receive", CONSUMER)
.setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "rocketmq")
.setAttribute(SemanticAttributes.MESSAGING_OPERATION, "receive");
Context rootContext = parentContext.with(spanBuilder.startSpan());
Context rootContext = withConsumerSpan(parentContext, spanBuilder.startSpan());
for (MessageExt message : msgs) {
createChildSpan(rootContext, message);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
apply from: "$rootDir/gradle/instrumentation.gradle"

ext {
// context "leak" here is intentional: spring-integration instrumentation will always override
// "local" span context with one extracted from the incoming message when it decides to start a
// CONSUMER span
failOnContextLeak = false
}
Comment on lines +3 to +8

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

This is a consequence of always using the context extracted from the incoming message; we completely ignore the "local" current context, but that's fine in that case.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Hmm... our debugContextLeakIfEnabled should not be called at all in case of CONSUMER spans. This is not spring integration specific problem.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

It's called for all spans, span kind does not matter.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Yes, but the logic of not failing the test should be the same for all CONSUMER spans.


muzzle {
pass {
group = "org.springframework.integration"
module = "spring-integration-core"
versions = "[4.1.0.RELEASE,)"
assertInverse = true
}
}

dependencies {
implementation project(':instrumentation:spring:spring-integration-4.1:library')

library 'org.springframework.integration:spring-integration-core:4.1.0.RELEASE'

testInstrumentation project(':instrumentation:rabbitmq-2.7:javaagent')

testImplementation project(':instrumentation:spring:spring-integration-4.1:testing')

testLibrary "org.springframework.boot:spring-boot-starter-test:1.5.22.RELEASE"
testLibrary "org.springframework.boot:spring-boot-starter:1.5.22.RELEASE"
testLibrary "org.springframework.cloud:spring-cloud-stream:2.2.1.RELEASE"
testLibrary "org.springframework.cloud:spring-cloud-stream-binder-rabbit:2.2.1.RELEASE"

testImplementation "javax.servlet:javax.servlet-api:3.1.0"
}

test {
filter {
excludeTestsMatching 'SpringIntegrationAndRabbitTest'
}
jvmArgs "-Dotel.instrumentation.rabbitmq.enabled=false"
}
test.finalizedBy(tasks.register("testWithRabbitInstrumentation", Test) {
filter {
includeTestsMatching 'SpringIntegrationAndRabbitTest'
}
jvmArgs "-Dotel.instrumentation.rabbitmq.enabled=true"
})

tasks.withType(Test).configureEach {
systemProperty "testLatestDeps", testLatestDeps
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
/*
* Copyright The OpenTelemetry Authors
* SPDX-License-Identifier: Apache-2.0
*/

package io.opentelemetry.javaagent.instrumentation.spring.integration;

import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.extendsClass;
import static io.opentelemetry.javaagent.extension.matcher.ClassLoaderMatcher.hasClassesNamed;
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
import static org.springframework.beans.factory.support.BeanDefinitionBuilder.genericBeanDefinition;

import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
import net.bytebuddy.asm.Advice;
import net.bytebuddy.description.type.TypeDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper;

public class ApplicationContextInstrumentation implements TypeInstrumentation {
@Override
public ElementMatcher<ClassLoader> classLoaderOptimization() {
return hasClassesNamed("org.springframework.context.support.AbstractApplicationContext");
}

@Override
public ElementMatcher<TypeDescription> typeMatcher() {
return extendsClass(named("org.springframework.context.support.AbstractApplicationContext"));
}

@Override
public void transform(TypeTransformer transformer) {
transformer.applyAdviceToMethod(
isMethod()
.and(named("postProcessBeanFactory"))
.and(
takesArgument(
0,
named(
"org.springframework.beans.factory.config.ConfigurableListableBeanFactory"))),
ApplicationContextInstrumentation.class.getName() + "$PostProcessBeanFactoryAdvice");
}

public static class PostProcessBeanFactoryAdvice {
@Advice.OnMethodEnter(suppress = Throwable.class)
public static void onEnter(@Advice.Argument(0) ConfigurableListableBeanFactory beanFactory) {
if (beanFactory instanceof BeanDefinitionRegistry
&& !beanFactory.containsBean("otelGlobalChannelInterceptor")) {

BeanDefinition globalChannelInterceptorBean =
genericBeanDefinition(GlobalChannelInterceptorWrapper.class)
.addConstructorArgValue(SpringIntegrationSingletons.interceptor())
.addPropertyValue("patterns", SpringIntegrationSingletons.patterns())
.getBeanDefinition();

((BeanDefinitionRegistry) beanFactory)
.registerBeanDefinition("otelGlobalChannelInterceptor", globalChannelInterceptorBean);
}
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
/*
* Copyright The OpenTelemetry Authors
* SPDX-License-Identifier: Apache-2.0
*/

package io.opentelemetry.javaagent.instrumentation.spring.integration;

import static java.util.Collections.singletonList;

import com.google.auto.service.AutoService;
import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
import java.util.List;

@AutoService(InstrumentationModule.class)
public class SpringIntegrationInstrumentationModule extends InstrumentationModule {
public SpringIntegrationInstrumentationModule() {
super("spring-integration", "spring-integration-4.1");
}

@Override
public List<TypeInstrumentation> typeInstrumentations() {
return singletonList(new ApplicationContextInstrumentation());
}
}
Loading