diff --git a/CHANGES.md b/CHANGES.md index 2dd610a5cd..706a53581f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -5,6 +5,13 @@ Release Notes. 9.8.0 ------------------ +* Fix the `spring-cloud-gateway-4.x` plugin propagating another request's context. Since Spring Cloud + Gateway 4.1.2 the outbound chain is assembled in `NettyRoutingFilter#filter` but subscribed later, so the + plugin parked the request's `ContextSnapshot` on the `HttpClient` returned by + `NettyRoutingFilter#getHttpClient`, which is the single shared bean unless a connect timeout is configured. + A concurrent request could overwrite it before the chain was subscribed, and the outbound `sw8` header then + carried a context the downstream service joined by mistake. The snapshot is held by a client derived per + request now (apache/skywalking#14095). * Fix `jedis-4.x-plugin`'s `AbstractConnectionInterceptor` double-stopping the span stack on any Redis-level exception (or a null dynamic field on a pooled/recycled `Connection`), which corrupted the parent trace for the rest of the request (apache/skywalking#14085). diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412Interceptor.java b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412Interceptor.java index 282e6f5f17..8af31a7780 100644 --- a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412Interceptor.java +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412Interceptor.java @@ -19,14 +19,19 @@ package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v412x; import org.apache.skywalking.apm.agent.core.context.ContextManager; -import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; import org.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x.define.EnhanceObjectCache; +import reactor.netty.http.client.HttpClient; import java.lang.reflect.Method; +/** + * Attach the context snapshot of the current request to the {@link HttpClient} returned by + * NettyRoutingFilter#getHttpClient, so that the outbound span created when the reactive chain is + * subscribed can continue the trace of the request that assembled it. + */ public class NettyRoutingGetHttpClientV412Interceptor implements InstanceMethodsAroundInterceptor { @Override @@ -36,16 +41,29 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { - if (ret instanceof EnhancedInstance) { - if (ContextManager.isActive()) { - ContextSnapshot contextSnapshot = ContextManager.capture(); - EnhanceObjectCache retEnhanceObjectCache = new EnhanceObjectCache(); - retEnhanceObjectCache.setContextSnapshot(contextSnapshot); - EnhancedInstance retEnhancedInstance = (EnhancedInstance) ret; - retEnhancedInstance.setSkyWalkingDynamicField(retEnhanceObjectCache); - } + if (!ContextManager.isActive() || !(ret instanceof HttpClient)) { + return ret; + } + /* + * NettyRoutingFilter#getHttpClient returns the shared HttpClient bean itself unless a connect timeout is + * configured, so the returned instance is the very same object for every request of the whole JVM. The + * snapshot is not read here but later, when the chain assembled by NettyRoutingFilter#filter is subscribed, + * therefore a concurrent request is able to overwrite it in between and the outbound sw8 header would carry + * another request's context. + * + * Derive a per-request client to hold the snapshot instead. HttpClient#headers duplicates the client and + * copies the header map only, which Spring Cloud Gateway copies once more right after, so nothing but one + * duplication is added to a chain that already duplicates on headers(), request() and uri(). + */ + final HttpClient perRequestHttpClient = ((HttpClient) ret).headers(headers -> { + }); + if (!(perRequestHttpClient instanceof EnhancedInstance)) { + return ret; } - return ret; + final EnhanceObjectCache enhanceObjectCache = new EnhanceObjectCache(); + enhanceObjectCache.setContextSnapshot(ContextManager.capture()); + ((EnhancedInstance) perRequestHttpClient).setSkyWalkingDynamicField(enhanceObjectCache); + return perRequestHttpClient; } @Override diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412InterceptorTest.java b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412InterceptorTest.java index 5f3f8811ae..010b14d3b3 100644 --- a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412InterceptorTest.java +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v412x/NettyRoutingGetHttpClientV412InterceptorTest.java @@ -19,9 +19,9 @@ package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v412x; import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; -import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule; import org.apache.skywalking.apm.agent.test.tools.SegmentStorage; @@ -29,48 +29,35 @@ import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; import org.apache.skywalking.apm.plugin.spring.cloud.gateway.v4x.define.EnhanceObjectCache; -import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.mockito.junit.MockitoJUnit; import org.mockito.junit.MockitoRule; -import java.util.List; +import reactor.netty.http.client.HttpClient; -import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; @RunWith(TracingSegmentRunner.class) public class NettyRoutingGetHttpClientV412InterceptorTest { private final static String ENTRY_OPERATION_NAME = "/get"; - private final NettyRoutingGetHttpClientV412Interceptor interceptor = new NettyRoutingGetHttpClientV412Interceptor(); - private final EnhancedInstance enhancedInstance = new EnhancedInstance() { - private EnhanceObjectCache enhanceObjectCache; - - @Override - public Object getSkyWalkingDynamicField() { - return enhanceObjectCache; - } - @Override - public void setSkyWalkingDynamicField(Object value) { - this.enhanceObjectCache = (EnhanceObjectCache) value; - } - }; - private final EnhancedInstance retEnhancedInstance = new EnhancedInstance() { - private EnhanceObjectCache enhanceObjectCache; - @Override - public Object getSkyWalkingDynamicField() { - return enhanceObjectCache; - } + private final NettyRoutingGetHttpClientV412Interceptor interceptor = new NettyRoutingGetHttpClientV412Interceptor(); - @Override - public void setSkyWalkingDynamicField(Object value) { - this.enhanceObjectCache = (EnhanceObjectCache) value; - } - }; + /** + * Stands for the single NettyRoutingFilter#httpClient bean, which + * NettyRoutingFilter#getHttpClient returns to every request. + */ + private final HttpClient sharedHttpClient = mockHttpClient(); @Rule public AgentServiceRule serviceRule = new AgentServiceRule(); @@ -79,37 +66,66 @@ public void setSkyWalkingDynamicField(Object value) { @SegmentStoragePoint private SegmentStorage segmentStorage; - private AbstractSpan entrySpan; - @Before - public void setUp() throws Exception { + @Test + public void testSnapshotIsHeldByADerivedClientAndNotByTheSharedOne() throws Throwable { + final HttpClient derivedHttpClient = mockHttpClient(); + when(sharedHttpClient.headers(any())).thenReturn(derivedHttpClient); + + final Object ret = getHttpClientWithinARequest(sharedHttpClient); + + assertSame(derivedHttpClient, ret); + assertNotNull(snapshotOf(derivedHttpClient)); + // The shared bean is reused by every request, so it must never hold a request scoped snapshot. + verify((EnhancedInstance) sharedHttpClient, never()).setSkyWalkingDynamicField(any()); } @Test - public void testWithContextIsActive() throws Throwable { - entrySpan = ContextManager.createEntrySpan(ENTRY_OPERATION_NAME, null); - entrySpan.setLayer(SpanLayer.HTTP); - entrySpan.setComponent(ComponentsDefine.SPRING_WEBFLUX); - interceptor.afterMethod(enhancedInstance, null, null, null, retEnhancedInstance); - assertNotNull(retEnhancedInstance.getSkyWalkingDynamicField()); - assertTrue(retEnhancedInstance.getSkyWalkingDynamicField() instanceof EnhanceObjectCache); - EnhanceObjectCache enhanceObjectCache = (EnhanceObjectCache) retEnhancedInstance.getSkyWalkingDynamicField(); - assertNotNull(enhanceObjectCache.getContextSnapshot()); - final List traceSegments = segmentStorage.getTraceSegments(); - assertEquals(traceSegments.size(), 0); - if (ContextManager.isActive()) { - ContextManager.stopSpan(); - } + public void testConcurrentRequestsDoNotShareTheSnapshot() throws Throwable { + final HttpClient firstDerivedHttpClient = mockHttpClient(); + final HttpClient secondDerivedHttpClient = mockHttpClient(); + when(sharedHttpClient.headers(any())).thenReturn(firstDerivedHttpClient, secondDerivedHttpClient); + + final Object firstRet = getHttpClientWithinARequest(sharedHttpClient); + final Object secondRet = getHttpClientWithinARequest(sharedHttpClient); + + assertSame(firstDerivedHttpClient, firstRet); + assertSame(secondDerivedHttpClient, secondRet); + assertNotEquals( + snapshotOf(firstDerivedHttpClient).getTraceId().getId(), + snapshotOf(secondDerivedHttpClient).getTraceId().getId() + ); + verify((EnhancedInstance) sharedHttpClient, never()).setSkyWalkingDynamicField(any()); } @Test public void testWithContextNotActive() throws Throwable { - interceptor.afterMethod(enhancedInstance, null, null, null, retEnhancedInstance); - assertNull(retEnhancedInstance.getSkyWalkingDynamicField()); - final List traceSegments = segmentStorage.getTraceSegments(); - assertEquals(traceSegments.size(), 0); - if (ContextManager.isActive()) { + final Object ret = interceptor.afterMethod(null, null, null, null, sharedHttpClient); + + assertSame(sharedHttpClient, ret); + // Nothing to propagate, so not even a client is derived. + verify(sharedHttpClient, never()).headers(any()); + verify((EnhancedInstance) sharedHttpClient, never()).setSkyWalkingDynamicField(any()); + } + + private Object getHttpClientWithinARequest(final HttpClient httpClient) throws Throwable { + final AbstractSpan entrySpan = ContextManager.createEntrySpan(ENTRY_OPERATION_NAME, null); + entrySpan.setLayer(SpanLayer.HTTP); + entrySpan.setComponent(ComponentsDefine.SPRING_WEBFLUX); + try { + return interceptor.afterMethod(null, null, null, null, httpClient); + } finally { ContextManager.stopSpan(); } } + + private ContextSnapshot snapshotOf(final HttpClient httpClient) { + final ArgumentCaptor captor = ArgumentCaptor.forClass(Object.class); + verify((EnhancedInstance) httpClient).setSkyWalkingDynamicField(captor.capture()); + return ((EnhanceObjectCache) captor.getValue()).getContextSnapshot(); + } + + private static HttpClient mockHttpClient() { + return mock(HttpClient.class, withSettings().extraInterfaces(EnhancedInstance.class)); + } }