add support for HttpExchange along with webflux-webclient-6.x

.github/workflows/plugins-jdk17-test.1.yaml

@@ -58,6 +58,7 @@ jobs:
  - spring-6.x-scenario
  - resteasy-6.x-scenario
  - gateway-4.x-scenario
+ - httpexchange-scenario
  steps:
  - uses: actions/checkout@v2
  with:

CHANGES.md

@@ -10,6 +10,7 @@ Release Notes.
 * Support for tracing spring-cloud-gateway 4.x in gateway-4.x-plugin.
 * Fix re-transform bug when plugin enhanced class proxy parent method.
 * Fix error HTTP status codes not recording as SLA failures in Vert.x plugins. 
+* Support for HttpExchange request tracing
 
 #### Documentation
 

apm-sniffer/apm-sdk-plugin/spring-plugins/pom.xml

@@ -43,6 +43,7 @@
  <module>spring-kafka-2.x-plugin</module>
  <module>scheduled-annotation-plugin</module>
  <module>spring-webflux-5.x-webclient-plugin</module>
+ <module>spring-webflux-6.x-webclient-plugin</module>
  <module>resttemplate-commons</module>
  </modules>
  <packaging>pom</packaging>

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-5.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/webclient/define/WebFluxWebClientInstrumentation.java

@@ -18,8 +18,11 @@
 
 package org.apache.skywalking.apm.plugin.spring.webflux.v5.webclient.define;
 
+import java.util.Collections;
+import java.util.List;
 import net.bytebuddy.description.method.MethodDescription;
 import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.WitnessMethod;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.v2.InstanceMethodsInterceptV2Point;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.StaticMethodsInterceptPoint;
@@ -32,6 +35,8 @@
 public class WebFluxWebClientInstrumentation extends ClassInstanceMethodsEnhancePluginDefineV2 {
  private static final String ENHANCE_CLASS = "org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction";
  private static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.spring.webflux.v5.webclient.WebFluxWebClientInterceptor";
+ private static final String WEBFLUX_CONTEXT_WRITE_CLASS = "reactor.core.publisher.Mono";
+ private static final String WEBFLUX_CONTEXT_WRITE_METHOD = "subscriberContext";
 
  @Override
  protected ClassMatch enhanceClass() {
@@ -69,4 +74,10 @@ public boolean isOverrideArgs() {
  public StaticMethodsInterceptPoint[] getStaticMethodsInterceptPoints() {
  return new StaticMethodsInterceptPoint[0];
  }
+
+ @Override
+ protected List<WitnessMethod> witnessMethods() {
+ return Collections.singletonList(
+ new WitnessMethod(WEBFLUX_CONTEXT_WRITE_CLASS, named(WEBFLUX_CONTEXT_WRITE_METHOD)));
+ }
 }

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-6.x-webclient-plugin/pom.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ ~
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>spring-plugins</artifactId>
+ <groupId>org.apache.skywalking</groupId>
+ <version>9.2.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>spring-webflux-6.x-webclient-plugin</artifactId>
+
+ <url>http://maven.apache.org</url>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-webflux</artifactId>
+ <version>6.0.0</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+</project>

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-6.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v6/webclient/BodyInserterRequestInterceptor.java

@@ -0,0 +1,54 @@
+/*
+ * 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.webflux.v6.webclient;
+
+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.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.springframework.http.client.reactive.ClientHttpRequest;
+
+import java.lang.reflect.Method;
+
+public class BodyInserterRequestInterceptor implements InstanceMethodsAroundInterceptor {
+
+ @Override
+ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
+ MethodInterceptResult result) throws Throwable {
+ ClientHttpRequest clientHttpRequest = (ClientHttpRequest) allArguments[0];
+ ContextCarrier contextCarrier = (ContextCarrier) objInst.getSkyWalkingDynamicField();
+ CarrierItem next = contextCarrier.items();
+ while (next.hasNext()) {
+ next = next.next();
+ clientHttpRequest.getHeaders().set(next.getHeadKey(), next.getHeadValue());
+ }
+ }
+
+ @Override
+ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
+ Object ret) throws Throwable {
+ return ret;
+ }
+
+ @Override
+ public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+ Class<?>[] argumentsTypes, Throwable t) {
+ }
+}

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-6.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v6/webclient/WebFluxWebClientInterceptor.java

@@ -0,0 +1,126 @@
+/*
+ * 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.webflux.v6.webclient;
+
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+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.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.v2.InstanceMethodsAroundInterceptorV2;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.MethodInvocationContext;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+import org.springframework.http.HttpStatusCode;
+import org.springframework.web.reactive.function.client.ClientRequest;
+import org.springframework.web.reactive.function.client.ClientResponse;
+import reactor.core.publisher.Mono;
+
+import java.lang.reflect.Method;
+import java.net.URI;
+import java.util.Optional;
+
+public class WebFluxWebClientInterceptor implements InstanceMethodsAroundInterceptorV2 {
+
+ @Override
+ public void beforeMethod(EnhancedInstance objInst,
+ Method method,
+ Object[] allArguments,
+ Class<?>[] argumentsTypes,
+ MethodInvocationContext context) throws Throwable {
+
+ }
+
+ @Override
+ public Object afterMethod(EnhancedInstance objInst,
+ Method method,
+ Object[] allArguments,
+ Class<?>[] argumentsTypes,
+ Object ret,
+ MethodInvocationContext context) throws Throwable {
+ // fix the problem that allArgument[0] may be null
+ if (allArguments[0] == null) {
+ return ret;
+ }
+ Mono<ClientResponse> ret1 = (Mono<ClientResponse>) ret;
+ return Mono.deferContextual(ctx -> {
+
+ ClientRequest request = (ClientRequest) allArguments[0];
+ URI uri = request.url();
+ final String operationName = getRequestURIString(uri);
+ final String remotePeer = getIPAndPort(uri);
+ AbstractSpan span = ContextManager.createExitSpan(operationName, remotePeer);
+
+ // get ContextSnapshot from reactor context, the snapshot is set to reactor context by any other plugin
+ // such as DispatcherHandlerHandleMethodInterceptor in spring-webflux-5.x-plugin
+ final Optional<Object> optional = ctx.getOrEmpty("SKYWALKING_CONTEXT_SNAPSHOT");
+ optional.ifPresent(snapshot -> ContextManager.continued((ContextSnapshot) snapshot));
+
+ //set components name
+ span.setComponent(ComponentsDefine.SPRING_WEBCLIENT);
+ Tags.URL.set(span, uri.toString());
+ Tags.HTTP.METHOD.set(span, request.method().toString());
+ SpanLayer.asHttp(span);
+
+ final ContextCarrier contextCarrier = new ContextCarrier();
+ ContextManager.inject(contextCarrier);
+ if (request instanceof EnhancedInstance) {
+ ((EnhancedInstance) request).setSkyWalkingDynamicField(contextCarrier);
+ }
+
+ //user async interface
+ span.prepareForAsync();
+ ContextManager.stopSpan();
+ return ret1.doOnSuccess(clientResponse -> {
+ HttpStatusCode httpStatus = clientResponse.statusCode();
+ if (httpStatus != null) {
+ Tags.HTTP_RESPONSE_STATUS_CODE.set(span, httpStatus.value());
+ if (httpStatus.isError()) {
+ span.errorOccurred();
+ }
+ }
+ }).doOnError(span::log).doFinally(s -> {
+ span.asyncFinish();
+ });
+ });
+ }
+
+ @Override
+ public void handleMethodException(EnhancedInstance objInst,
+ Method method,
+ Object[] allArguments,
+ Class<?>[] argumentsTypes,
+ Throwable t,
+ MethodInvocationContext context) {
+ AbstractSpan activeSpan = ContextManager.activeSpan();
+ activeSpan.errorOccurred();
+ activeSpan.log(t);
+ }
+
+ private String getRequestURIString(URI uri) {
+ String requestPath = uri.getPath();
+ return requestPath != null && requestPath.length() > 0 ? requestPath : "/";
+ }
+
+ // return ip:port
+ private String getIPAndPort(URI uri) {
+ return uri.getHost() + ":" + uri.getPort();
+ }
+}

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-6.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v6/webclient/define/BodyInserterRequestInstrumentation.java

@@ -0,0 +1,67 @@
+/*
+ * 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.webflux.v6.webclient.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.enhance.ClassInstanceMethodsEnhancePluginDefine;
+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.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+public class BodyInserterRequestInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+ @Override
+ public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+ return new ConstructorInterceptPoint[0];
+ }
+
+ @Override
+ public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+ return new InstanceMethodsInterceptPoint[] {
+ new InstanceMethodsInterceptPoint() {
+ @Override
+ public ElementMatcher<MethodDescription> getMethodsMatcher() {
+ return named("writeTo").and(
+ takesArgumentWithType(0, "org.springframework.http.client.reactive.ClientHttpRequest"));
+ }
+
+ @Override
+ public String getMethodsInterceptor() {
+ return "org.apache.skywalking.apm.plugin.spring.webflux.v6.webclient.BodyInserterRequestInterceptor";
+ }
+
+ @Override
+ public boolean isOverrideArgs() {
+ return false;
+ }
+ }
+ };
+ }
+
+ @Override
+ protected ClassMatch enhanceClass() {
+ return byName(
+ "org.springframework.web.reactive.function.client.DefaultClientRequestBuilder$BodyInserterRequest");
+ }
+}