Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9001: Revise Observation propagation over the channel

Browse Source 
Fixes: #9001

The `ObservationPropagationChannelInterceptor` does not propagate an observation properly.
And it fully cannot when the message channel is persistent.

* Deprecate `ObservationPropagationChannelInterceptor` in favor of enabled observation on the channel
and target `MessageHandler` which is a consumer of this channel.
* Remove tests with an `ObservationPropagationChannelInterceptor`
* Mention a correct behavior in the `metrics.adoc` and `ObservationPropagationChannelInterceptor` Javadocs
This commit is contained in:
Artem Bilan
2024-03-18 16:19:00 -04:00
parent 91d21c2477
commit 0b1d69191f
5 changed files with 18 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-webflux/src/test/java/org/springframework/integration/webflux/observation/WebFluxObservationPropagationTests.java
View File

@@ -23,7 +23,6 @@ import brave.propagation.ThreadLocalCurrentTraceContext;
import brave.test.TestSpanHandler;
import io.micrometer.observation.ObservationHandler;
import io.micrometer.observation.ObservationRegistry;
import io.micrometer.observation.tck.TestObservationRegistryAssert;
import io.micrometer.tracing.Tracer;
import io.micrometer.tracing.brave.bridge.BraveBaggageManager;
import io.micrometer.tracing.brave.bridge.BraveCurrentTraceContext;
@@ -46,15 +45,12 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpMethod;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.interceptor.ObservationPropagationChannelInterceptor;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableIntegrationManagement;
import org.springframework.integration.config.GlobalChannelInterceptor;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.webflux.dsl.WebFlux;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.web.SpringJUnitWebConfig;
import org.springframework.test.web.reactive.server.HttpHandlerConnector;
@@ -104,12 +100,8 @@ public class WebFluxObservationPropagationTests {
.extracting(Message::getPayload)
.isEqualTo("Received data: " + testData);
TestObservationRegistryAssert.assertThat(this.observationRegistry)
.hasRemainingCurrentObservation();
this.observationRegistry.getCurrentObservation().stop();
assertThat(SPANS.spans()).hasSize(6);
// There is a race condition when we already have a reply, but the span in the last channel is not closed yet.
await().untilAsserted(() -> assertThat(SPANS.spans()).hasSize(6));
SpansAssert.assertThat(SPANS.spans().stream().map(BraveFinishedSpan::fromBrave).collect(Collectors.toList()))
.haveSameTraceId();
}
@@ -179,12 +171,6 @@ public class WebFluxObservationPropagationTests {
return WebTestClient.bindToServer(new HttpHandlerConnector(httpHandler)).build();
}
@Bean
@GlobalChannelInterceptor
public ChannelInterceptor observationPropagationInterceptor(ObservationRegistry observationRegistry) {
return new ObservationPropagationChannelInterceptor(observationRegistry);
}
@Bean
IntegrationFlow webFluxFlow() {
return IntegrationFlow