Test spring-integration instrumentation with spring-cloud-stream example
MODIFY
instrumentation/spring/spring-integration-4.1/library/spring-integration-4.1-library.gradle
instrumentation/spring/spring-integration-4.1/library/spring-integration-4.1-library.gradle
@@ -6,8 +6,13 @@ dependencies {
library 'org.springframework.integration:spring-integration-core:4.1.0.RELEASE'
+ testImplementation "org.testcontainers:testcontainers"
+
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"
}
test {

MODIFY
instrumentation/spring/spring-integration-4.1/library/src/test/groovy/ComplexPropagationTest.groovy
instrumentation/spring/spring-integration-4.1/library/src/test/groovy/ComplexPropagationTest.groovy
import static io.opentelemetry.instrumentation.test.utils.TraceUtils.runUnderTrace
-import io.opentelemetry.api.GlobalOpenTelemetry
-import io.opentelemetry.instrumentation.spring.integration.SpringIntegrationTracing
import io.opentelemetry.instrumentation.test.LibraryInstrumentationSpecification
import java.util.concurrent.BlockingQueue
import java.util.concurrent.ExecutorService
@@ -22,10 +20,8 @@ import org.springframework.context.annotation.Bean
import org.springframework.context.event.EventListener
import org.springframework.integration.channel.DirectChannel
import org.springframework.integration.channel.ExecutorChannel
-import org.springframework.integration.config.GlobalChannelInterceptor
import org.springframework.messaging.Message
import org.springframework.messaging.SubscribableChannel
-import org.springframework.messaging.support.ChannelInterceptor
import org.springframework.messaging.support.MessageBuilder
import spock.lang.Shared
@@ -35,7 +31,10 @@ class ComplexPropagationTest extends LibraryInstrumentationSpecification {
ConfigurableApplicationContext applicationContext
def setupSpec() {
- def app = new SpringApplication(ExternalQueueConfig)
+ def app = new SpringApplication(ExternalQueueConfig, GlobalInterceptorSpringConfig)
+ app.setDefaultProperties([
+ "spring.main.web-application-type": "none"
+ ])
applicationContext = app.run()
}
@@ -122,12 +121,6 @@ class ComplexPropagationTest extends LibraryInstrumentationSpecification {
}
})
}
-
- @GlobalChannelInterceptor
- @Bean
- ChannelInterceptor otelInterceptor() {
- SpringIntegrationTracing.create(GlobalOpenTelemetry.get()).newChannelInterceptor()
- }
}
static class Payload {

ADD
instrumentation/spring/spring-integration-4.1/library/src/test/groovy/GlobalInterceptorSpringConfig.groovy
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import io.opentelemetry.api.GlobalOpenTelemetry
+import io.opentelemetry.instrumentation.spring.integration.SpringIntegrationTracing
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.integration.config.GlobalChannelInterceptor
+import org.springframework.messaging.support.ChannelInterceptor
+
+@Configuration
+class GlobalInterceptorSpringConfig {
+
+ @GlobalChannelInterceptor
+ @Bean
+ ChannelInterceptor otelInterceptor() {
+ SpringIntegrationTracing.create(GlobalOpenTelemetry.get()).newChannelInterceptor()
+ }
+}

ADD
instrumentation/spring/spring-integration-4.1/library/src/test/groovy/SpringCloudStreamRabbitTest.groovy
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import static io.opentelemetry.instrumentation.test.utils.TraceUtils.runInternalSpan
+import static io.opentelemetry.instrumentation.test.utils.TraceUtils.runUnderTrace
+
+import io.opentelemetry.instrumentation.test.LibraryInstrumentationSpecification
+import java.time.Duration
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.SpringApplication
+import org.springframework.boot.SpringBootConfiguration
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration
+import org.springframework.cloud.stream.annotation.EnableBinding
+import org.springframework.cloud.stream.annotation.StreamListener
+import org.springframework.cloud.stream.messaging.Sink
+import org.springframework.cloud.stream.messaging.Source
+import org.springframework.context.ConfigurableApplicationContext
+import org.springframework.context.annotation.Bean
+import org.springframework.messaging.support.MessageBuilder
+import org.testcontainers.containers.GenericContainer
+import spock.lang.Shared
+
+class SpringCloudStreamRabbitTest extends LibraryInstrumentationSpecification {
+
+ @Shared
+ def rabbitMQContainer
+
+ @Shared
+ ConfigurableApplicationContext producerContext
+ @Shared
+ ConfigurableApplicationContext consumerContext
+
+ def setupSpec() {
+ rabbitMQContainer = new GenericContainer('rabbitmq:latest')
+ .withExposedPorts(5672)
+ .withStartupTimeout(Duration.ofSeconds(120))
+ rabbitMQContainer.start()
+
+ def producerApp = new SpringApplication(ProducerConfig, GlobalInterceptorSpringConfig)
+ producerApp.setDefaultProperties([
+ "spring.application.name" : "testProducer",
+ "spring.jmx.enabled" : false,
+ "spring.main.web-application-type" : "none",
+ "spring.rabbitmq.host" : rabbitMQContainer.containerIpAddress,
+ "spring.rabbitmq.port" : rabbitMQContainer.getMappedPort(5672),
+ "spring.cloud.stream.bindings.output.destination": "testTopic"
+ ])
+ producerContext = producerApp.run()
+
+ def consumerApp = new SpringApplication(ConsumerConfig, GlobalInterceptorSpringConfig)
+ consumerApp.setDefaultProperties([
+ "spring.application.name" : "testConsumer",
+ "spring.jmx.enabled" : false,
+ "spring.main.web-application-type" : "none",
+ "spring.rabbitmq.host" : rabbitMQContainer.containerIpAddress,
+ "spring.rabbitmq.port" : rabbitMQContainer.getMappedPort(5672),
+ "spring.cloud.stream.bindings.input.destination": "testTopic"
+ ])
+ consumerContext = consumerApp.run()
+ }
+
+ def cleanupSpec() {
+ rabbitMQContainer?.stop()
+ producerContext?.close()
+ consumerContext?.close()
+ }
+
+ def "should propagate context through RabbitMQ"() {
+ when:
+ producerContext.getBean("producer", Runnable).run()
+
+ then:
+ assertTraces(1) {
+ trace(0, 4) {
+ span(0) {
+ name "producer"
+ }
+ span(1) {
+ name "testProducer.output"
+ childOf span(0)
+ }
+ span(2) {
+ name "testConsumer.input"
+ childOf span(1)
+ }
+ span(3) {
+ name "consumer"
+ childOf span(2)
+ }
+ }
+ }
+ }
+
+ @SpringBootConfiguration
+ @EnableAutoConfiguration
+ @EnableBinding(Source)
+ static class ProducerConfig {
+ @Autowired
+ Source source
+
+ @Bean
+ Runnable producer() {
+ return {
+ runUnderTrace("producer") {
+ source.output().send(MessageBuilder.withPayload("test").build())
+ }
+ }
+ }
+ }
+
+ @SpringBootConfiguration
+ @EnableAutoConfiguration
+ @EnableBinding(Sink)
+ static class ConsumerConfig {
+ @StreamListener(Sink.INPUT)
+ void consume(String ignored) {
+ runInternalSpan("consumer")
+ }
+ }
+}

MODIFY
instrumentation/spring/spring-integration-4.1/library/src/test/groovy/SpringIntegrationTracingTest.groovy
instrumentation/spring/spring-integration-4.1/library/src/test/groovy/SpringIntegrationTracingTest.groovy
import static io.opentelemetry.instrumentation.test.utils.TraceUtils.runUnderTrace
-import io.opentelemetry.api.GlobalOpenTelemetry
-import io.opentelemetry.instrumentation.spring.integration.SpringIntegrationTracing
import io.opentelemetry.instrumentation.test.LibraryInstrumentationSpecification
import io.opentelemetry.sdk.trace.data.SpanData
import java.util.concurrent.Executors
@@ -18,7 +16,6 @@ import org.springframework.context.ConfigurableApplicationContext
import org.springframework.context.annotation.Bean
import org.springframework.context.event.EventListener
import org.springframework.integration.channel.DirectChannel
-import org.springframework.integration.config.GlobalChannelInterceptor
import org.springframework.messaging.Message
import org.springframework.messaging.SubscribableChannel
import org.springframework.messaging.support.ChannelInterceptor
@@ -34,7 +31,10 @@ class SpringIntegrationTracingTest extends LibraryInstrumentationSpecification {
ConfigurableApplicationContext applicationContext
def setupSpec() {
- def app = new SpringApplication(MessageChannelsConfig)
+ def app = new SpringApplication(MessageChannelsConfig, GlobalInterceptorSpringConfig)
+ app.setDefaultProperties([
+ "spring.main.web-application-type": "none"
+ ])
applicationContext = app.run()
}
@@ -147,12 +147,12 @@ class SpringIntegrationTracingTest extends LibraryInstrumentationSpecification {
}
@Bean
- SubscribableChannel executorChannel() {
+ SubscribableChannel executorChannel(ChannelInterceptor otelInterceptor) {
def channel = new ExecutorSubscribableChannel(Executors.newSingleThreadExecutor())
if (!Boolean.getBoolean("testLatestDeps")) {
// spring does not inject the interceptor in 4.1 because ExecutorSubscribableChannel isn't ChannelInterceptorAware
// in later versions spring injects the global interceptor into InterceptableChannel (which ExecutorSubscribableChannel is)
- channel.addInterceptor(otelInterceptor())
+ channel.addInterceptor(otelInterceptor)
}
channel
}
@@ -173,11 +173,5 @@ class SpringIntegrationTracingTest extends LibraryInstrumentationSpecification {
linkedChannel2().send(message)
}
}
-
- @GlobalChannelInterceptor
- @Bean
- ChannelInterceptor otelInterceptor() {
- SpringIntegrationTracing.create(GlobalOpenTelemetry.get()).newChannelInterceptor()
- }
}
}

Test spring-integration instrumentation with spring-cloud-stream example (#3241)