diff --git a/CHANGES.md b/CHANGES.md index ccff2d786..1e32c43a8 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -9,6 +9,7 @@ Release Notes. * Support Jetty 11.x plugin * Fix the scenario of using the HBase plugin with spring-data-hadoop. * Add RocketMQ 5.x plugin +* Fix the thread safety bug of finishing operation for the span named "SpringCloudGateway/sendRequest" #### Documentation diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/HttpClientRequestInterceptor.java b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/HttpClientRequestInterceptor.java index 00613b602..0bed218b2 100644 --- a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/HttpClientRequestInterceptor.java +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/HttpClientRequestInterceptor.java @@ -21,6 +21,7 @@ import java.lang.reflect.Method; import java.net.URL; import java.util.function.BiConsumer; import java.util.function.Function; + import org.apache.skywalking.apm.agent.core.context.CarrierItem; import org.apache.skywalking.apm.agent.core.context.ContextCarrier; import org.apache.skywalking.apm.agent.core.context.ContextManager; @@ -28,8 +29,8 @@ import org.apache.skywalking.apm.agent.core.context.tag.Tags; 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.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.agent.core.plugin.interceptor.enhance.v2.InstanceMethodsAroundInterceptorV2; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.MethodInvocationContext; import org.apache.skywalking.apm.plugin.spring.cloud.gateway.v20x.define.EnhanceCacheObject; import org.reactivestreams.Publisher; import reactor.core.publisher.Mono; @@ -38,13 +39,14 @@ import reactor.ipc.netty.http.client.HttpClientResponse; import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.SPRING_CLOUD_GATEWAY; -public class HttpClientRequestInterceptor implements InstanceMethodsAroundInterceptor { +public class HttpClientRequestInterceptor implements InstanceMethodsAroundInterceptorV2 { + @Override public void beforeMethod(final EnhancedInstance objInst, final Method method, final Object[] allArguments, final Class[] argumentsTypes, - final MethodInterceptResult result) throws Throwable { + final MethodInvocationContext context) throws Throwable { /* In this plug-in, the HttpClientFinalizerSendInterceptor depends on the NettyRoutingFilterInterceptor @@ -54,13 +56,13 @@ public class HttpClientRequestInterceptor implements InstanceMethodsAroundInterc if (!ContextManager.isActive()) { return; } - + AbstractSpan span = ContextManager.activeSpan(); URL url = new URL((String) allArguments[1]); ContextCarrier contextCarrier = new ContextCarrier(); AbstractSpan abstractSpan = ContextManager.createExitSpan( - "SpringCloudGateway/sendRequest", contextCarrier, getPeer(url)); + "SpringCloudGateway/sendRequest", contextCarrier, getPeer(url)); abstractSpan.prepareForAsync(); Tags.URL.set(abstractSpan, String.valueOf(allArguments[1])); abstractSpan.setLayer(SpanLayer.HTTP); @@ -80,7 +82,7 @@ public class HttpClientRequestInterceptor implements InstanceMethodsAroundInterc } }; - objInst.setSkyWalkingDynamicField(new EnhanceCacheObject(span, abstractSpan)); + context.setContext(new EnhanceCacheObject(span, abstractSpan)); } @Override @@ -88,30 +90,47 @@ public class HttpClientRequestInterceptor implements InstanceMethodsAroundInterc final Method method, final Object[] allArguments, final Class[] argumentsTypes, - final Object ret) { - EnhanceCacheObject enhanceCacheObject = (EnhanceCacheObject) objInst.getSkyWalkingDynamicField(); + final Object ret, + MethodInvocationContext context) { + EnhanceCacheObject enhanceCacheObject = (EnhanceCacheObject) context.getContext(); Mono responseMono = (Mono) ret; return responseMono.doAfterSuccessOrError(new BiConsumer() { @Override public void accept(final HttpClientResponse httpClientResponse, final Throwable throwable) { - - AbstractSpan abstractSpan = enhanceCacheObject.getSendSpan(); - if (abstractSpan != null) { - if (throwable != null) { - abstractSpan.log(throwable); - } else if (httpClientResponse.status().code() > 400) { - abstractSpan.errorOccurred(); - } - Tags.HTTP_RESPONSE_STATUS_CODE.set(abstractSpan, httpClientResponse.status().code()); - abstractSpan.asyncFinish(); - } - - objInst.setSkyWalkingDynamicField(null); - enhanceCacheObject.getFilterSpan().asyncFinish(); + doAfterSuccessOrError(httpClientResponse, throwable, enhanceCacheObject); } }); } + void doAfterSuccessOrError(HttpClientResponse httpClientResponse, Throwable throwable, EnhanceCacheObject enhanceCacheObject) { + try { + //When executing the beforeMethod method, if the ContextManager is inactive, the enhanceCacheObject will be null. + if (enhanceCacheObject == null) { + return; + } + + //The doAfterSuccessOrError method may be executed multiple times. + if (enhanceCacheObject.isSpanFinish()) { + return; + } + + AbstractSpan abstractSpan = enhanceCacheObject.getSendSpan(); + if (throwable != null) { + abstractSpan.log(throwable); + } else if (httpClientResponse.status().code() > 400) { + abstractSpan.errorOccurred(); + } + Tags.HTTP_RESPONSE_STATUS_CODE.set(abstractSpan, httpClientResponse.status().code()); + + abstractSpan.asyncFinish(); + enhanceCacheObject.getFilterSpan().asyncFinish(); + + enhanceCacheObject.setSpanFinish(true); + } catch (Throwable e) { + //Catch unknown exceptions to avoid interrupting business processes. + } + } + private String getPeer(URL url) { return url.getHost() + ":" + url.getPort(); } @@ -121,7 +140,8 @@ public class HttpClientRequestInterceptor implements InstanceMethodsAroundInterc final Method method, final Object[] allArguments, final Class[] argumentsTypes, - final Throwable t) { + final Throwable t, + MethodInvocationContext context) { } } diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/AbstractGateway200EnhancePluginDefineV2.java b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/AbstractGateway200EnhancePluginDefineV2.java new file mode 100644 index 000000000..1f07c0ba8 --- /dev/null +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/AbstractGateway200EnhancePluginDefineV2.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v20x.define; + +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.ClassInstanceMethodsEnhancePluginDefineV2; + +public abstract class AbstractGateway200EnhancePluginDefineV2 extends ClassInstanceMethodsEnhancePluginDefineV2 { + + @Override + protected String[] witnessClasses() { + return new String[] { + "org.springframework.cloud.gateway.config.GatewayAutoConfiguration$1" + }; + } +} diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/EnhanceCacheObject.java b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/EnhanceCacheObject.java index 14964a305..e5070ab7a 100644 --- a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/EnhanceCacheObject.java +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/EnhanceCacheObject.java @@ -22,6 +22,7 @@ import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; public class EnhanceCacheObject { private final AbstractSpan filterSpan; private final AbstractSpan sendSpan; + private volatile boolean spanFinish = false; public EnhanceCacheObject(final AbstractSpan filterSpan, final AbstractSpan sendSpan) { this.filterSpan = filterSpan; @@ -35,4 +36,12 @@ public class EnhanceCacheObject { public AbstractSpan getSendSpan() { return sendSpan; } + + public boolean isSpanFinish() { + return spanFinish; + } + + public void setSpanFinish(boolean spanFinish) { + this.spanFinish = spanFinish; + } } diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/HttpClientInstrumentation.java b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/HttpClientInstrumentation.java index 04031cfd9..b73eaca3f 100644 --- a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/HttpClientInstrumentation.java +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/define/HttpClientInstrumentation.java @@ -20,13 +20,13 @@ package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v20x.define; import net.bytebuddy.description.method.MethodDescription; import net.bytebuddy.matcher.ElementMatcher; import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; -import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.v2.InstanceMethodsInterceptV2Point; import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; import static net.bytebuddy.matcher.ElementMatchers.named; import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; -public class HttpClientInstrumentation extends AbstractGateway200EnhancePluginDefine { +public class HttpClientInstrumentation extends AbstractGateway200EnhancePluginDefineV2 { @Override protected ClassMatch enhanceClass() { @@ -39,16 +39,16 @@ public class HttpClientInstrumentation extends AbstractGateway200EnhancePluginDe } @Override - public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { - return new InstanceMethodsInterceptPoint[] { - new InstanceMethodsInterceptPoint() { + public InstanceMethodsInterceptV2Point[] getInstanceMethodsInterceptV2Points() { + return new InstanceMethodsInterceptV2Point[] { + new InstanceMethodsInterceptV2Point() { @Override public ElementMatcher getMethodsMatcher() { return named("request"); } @Override - public String getMethodsInterceptor() { + public String getMethodsInterceptorV2() { return Constants.REQUEST_INTERCEPTOR; } @@ -59,4 +59,5 @@ public class HttpClientInstrumentation extends AbstractGateway200EnhancePluginDe } }; } + } diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/HttpClientRequestInterceptorTest.java b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/HttpClientRequestInterceptorTest.java new file mode 100644 index 000000000..5ec786e6a --- /dev/null +++ b/apm-sniffer/optional-plugins/optional-spring-plugins/optional-spring-cloud/gateway-2.0.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/cloud/gateway/v20x/HttpClientRequestInterceptorTest.java @@ -0,0 +1,91 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.skywalking.apm.plugin.spring.cloud.gateway.v20x; + +import io.netty.handler.codec.http.HttpResponseStatus; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule; +import org.apache.skywalking.apm.agent.test.tools.SegmentStorage; +import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint; +import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.apache.skywalking.apm.plugin.spring.cloud.gateway.v20x.define.EnhanceCacheObject; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; +import reactor.ipc.netty.http.client.HttpClientResponse; + +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +@RunWith(TracingSegmentRunner.class) +public class HttpClientRequestInterceptorTest { + + private HttpClientRequestInterceptor httpClientRequestInterceptor = new HttpClientRequestInterceptor(); + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + @Rule + public MockitoRule rule = MockitoJUnit.rule(); + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + private HttpClientResponse httpClientResponse; + + @Before + public void setUp() throws Exception { + + httpClientResponse = Mockito.mock(HttpClientResponse.class); + HttpResponseStatus httpResponseStatus = Mockito.mock(HttpResponseStatus.class); + + Mockito.when(httpResponseStatus.code()).thenReturn(200); + Mockito.when(httpClientResponse.status()).thenReturn(httpResponseStatus); + } + + @Test + public void testDoAfterSuccessOrError() { + AbstractSpan filterSpan = ContextManager.createLocalSpan("mockFilterSpan"); + filterSpan.prepareForAsync(); + ContextManager.stopSpan(filterSpan); + + AbstractSpan sendSpan = ContextManager.createExitSpan("SpringCloudGateway/sendRequest", "http://127.0.0.1:80"); + sendSpan.prepareForAsync(); + ContextManager.stopSpan(sendSpan); + + EnhanceCacheObject enhanceCacheObject = new EnhanceCacheObject(filterSpan, sendSpan); + enhanceCacheObject = spy(enhanceCacheObject); + + //Test the ContextManager is inactive. + httpClientRequestInterceptor.doAfterSuccessOrError(httpClientResponse, null, null); + verify(enhanceCacheObject, Mockito.times(0)).setSpanFinish(true); + + //Test normal scenario. + httpClientRequestInterceptor.doAfterSuccessOrError(httpClientResponse, null, enhanceCacheObject); + verify(enhanceCacheObject, Mockito.times(1)).setSpanFinish(true); + + //Test the doAfterSuccessOrError method is executed multiple times. + httpClientRequestInterceptor.doAfterSuccessOrError(httpClientResponse, null, enhanceCacheObject); + verify(enhanceCacheObject, Mockito.times(1)).setSpanFinish(true); + } + +} \ No newline at end of file