diff --git a/.github/workflows/plugins-test.3.yaml b/.github/workflows/plugins-test.3.yaml index 2b262ea04..94b110bec 100644 --- a/.github/workflows/plugins-test.3.yaml +++ b/.github/workflows/plugins-test.3.yaml @@ -53,6 +53,8 @@ jobs: - { name: 'graphql-9.x-scenario', title: 'graphql-9.x 9.0-11.0 (3)' } - { name: 'graphql-12.x-scenario', title: 'graphql-12.x 12.0-15.0 (4)' } - { name: 'hbase-scenario', title: 'hbase-scenario (5)' } + - { name: 'spring-kafka-2.2.x-scenario', title: 'Spring-Kafka 2.2.x (7)' } + - { name: 'spring-kafka-2.3.x-scenario', title: 'Spring-Kafka 2.3.x (7)' } steps: - uses: actions/checkout@v2 with: diff --git a/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java index d46a61c84..1ba2bc3ff 100755 --- a/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java +++ b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java @@ -170,4 +170,6 @@ public class ComponentsDefine { public static final OfficialComponent SPRING_ANNOTATION = new OfficialComponent(93, "spring-annotation"); public static final OfficialComponent HBASE = new OfficialComponent(94, "HBase"); + + public static final OfficialComponent SPRING_KAFKA_CONSUMER = new OfficialComponent(95, "spring-kafka-consumer"); } diff --git a/apm-sniffer/apm-sdk-plugin/kafka-commons/pom.xml b/apm-sniffer/apm-sdk-plugin/kafka-commons/pom.xml new file mode 100644 index 000000000..450bbd196 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-commons/pom.xml @@ -0,0 +1,46 @@ + + + + + 4.0.0 + + + org.apache.skywalking + apm-sdk-plugin + 8.2.0-SNAPSHOT + + + apm-kafka-commons + kafka-commons + + + 2.0.1 + + + + + org.apache.kafka + kafka-clients + ${kafka-clients.version} + provided + + + \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapter.java b/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapter.java new file mode 100644 index 000000000..ac94208c8 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapter.java @@ -0,0 +1,62 @@ +/* + * 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.kafka; + +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; + +/** + * implements Callback and EnhancedInstance, for transformation kafkaTemplate.buildCallback + */ +public class CallbackAdapter implements Callback, EnhancedInstance { + + private Object instance; + + private Callback userCallback; + + public CallbackAdapter(Callback userCallback, Object instance) { + this.userCallback = userCallback; + this.instance = instance; + } + + public CallbackAdapter() { + } + + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + if (userCallback != null) { + userCallback.onCompletion(metadata, exception); + } + } + + @Override + public Object getSkyWalkingDynamicField() { + return instance; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + this.instance = value; + } + + public Callback getUserCallback() { + return userCallback; + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/AbstractKafkaTemplateInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/AbstractKafkaTemplateInstrumentation.java similarity index 100% rename from apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/AbstractKafkaTemplateInstrumentation.java rename to apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/AbstractKafkaTemplateInstrumentation.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Constants.java b/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Constants.java new file mode 100644 index 000000000..fd68eaa60 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Constants.java @@ -0,0 +1,27 @@ +/* + * 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.kafka.define; + +public class Constants { + + public static final String SPRING_KAFKA_FLAG = "SW_SPRING_KAFKA_FLAG"; + + public static final String SPRING_KAFKA_POLL_AND_INVOKE_OPERATION_NAME = "/spring-kafka/pollAndInvoke"; +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/SpringKafkaContext.java b/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/SpringKafkaContext.java new file mode 100644 index 000000000..e004423c7 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-commons/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/SpringKafkaContext.java @@ -0,0 +1,37 @@ +/* + * 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.kafka.define; + +public class SpringKafkaContext { + + public SpringKafkaContext() { + needStop = false; + } + + private boolean needStop; + + public boolean isNeedStop() { + return needStop; + } + + public void setNeedStop(boolean needStop) { + this.needStop = needStop; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/kafka-plugin/pom.xml index 2d0d97190..5fe80b061 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/pom.xml +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/pom.xml @@ -17,7 +17,8 @@ ~ --> - + apm-sdk-plugin org.apache.skywalking @@ -28,10 +29,16 @@ apm-kafka-plugin - 0.11.0.0 + 2.0.1 + + org.apache.skywalking + apm-kafka-commons + ${project.version} + provided + org.apache.kafka kafka-clients diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaConsumerInterceptor.java index 46e1ad933..6318456d1 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaConsumerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaConsumerInterceptor.java @@ -18,10 +18,6 @@ package org.apache.skywalking.apm.plugin.kafka; -import java.lang.reflect.Method; -import java.util.Iterator; -import java.util.List; -import java.util.Map; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Header; @@ -35,6 +31,13 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedI 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.network.trace.component.ComponentsDefine; +import org.apache.skywalking.apm.plugin.kafka.define.Constants; +import org.apache.skywalking.apm.plugin.kafka.define.SpringKafkaContext; + +import java.lang.reflect.Method; +import java.util.Iterator; +import java.util.List; +import java.util.Map; public class KafkaConsumerInterceptor implements InstanceMethodsAroundInterceptor { @@ -43,14 +46,14 @@ public class KafkaConsumerInterceptor implements InstanceMethodsAroundIntercepto @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - MethodInterceptResult result) throws Throwable { + MethodInterceptResult result) throws Throwable { ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField(); requiredInfo.setStartTime(System.currentTimeMillis()); } @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - Object ret) throws Throwable { + Object ret) throws Throwable { /* * If the intercepted method throws exception, the ret will be null */ @@ -63,8 +66,12 @@ public class KafkaConsumerInterceptor implements InstanceMethodsAroundIntercepto // if (records.size() > 0) { ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField(); - AbstractSpan activeSpan = ContextManager.createEntrySpan(OPERATE_NAME_PREFIX + requiredInfo.getTopics() + CONSUMER_OPERATE_NAME + requiredInfo - .getGroupId(), null).start(requiredInfo.getStartTime()); + if (ContextManager.getRuntimeContext().get(Constants.SPRING_KAFKA_FLAG) != null) { + ContextManager.createEntrySpan(Constants.SPRING_KAFKA_POLL_AND_INVOKE_OPERATION_NAME, null); + ((SpringKafkaContext) ContextManager.getRuntimeContext().get(Constants.SPRING_KAFKA_FLAG)).setNeedStop(true); + } + String operationName = OPERATE_NAME_PREFIX + requiredInfo.getTopics() + CONSUMER_OPERATE_NAME + requiredInfo.getGroupId(); + AbstractSpan activeSpan = ContextManager.createEntrySpan(operationName, null).start(requiredInfo.getStartTime()); activeSpan.setComponent(ComponentsDefine.KAFKA_CONSUMER); SpanLayer.asMQ(activeSpan); @@ -93,7 +100,7 @@ public class KafkaConsumerInterceptor implements InstanceMethodsAroundIntercepto @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, - Class[] argumentsTypes, Throwable t) { + Class[] argumentsTypes, Throwable t) { /* * The entry span is created in {@link #afterMethod}, but {@link #handleMethodException} is called before * {@link #afterMethod}, before the creation of entry span, we can not ensure there is an active span diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def index 64f3d72fc..d807aceef 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def @@ -18,5 +18,4 @@ kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.CallbackInstr kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaConsumerInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerMapInstrumentation -kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateCallbackInstrumentation \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/pom.xml index 1efd8d9ca..df55b991a 100644 --- a/apm-sniffer/apm-sdk-plugin/pom.xml +++ b/apm-sniffer/apm-sdk-plugin/pom.xml @@ -63,6 +63,7 @@ elastic-job-3.x-plugin mongodb-2.x-plugin httpasyncclient-4.x-plugin + kafka-commons kafka-plugin servicecomb-plugin hystrix-1.x-plugin diff --git a/apm-sniffer/apm-sdk-plugin/spring-plugins/pom.xml b/apm-sniffer/apm-sdk-plugin/spring-plugins/pom.xml index 0e5da3992..c3483080e 100644 --- a/apm-sniffer/apm-sdk-plugin/spring-plugins/pom.xml +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/pom.xml @@ -39,6 +39,7 @@ spring-commons mvc-annotation-5.x-plugin spring-webflux-5.x-plugin + spring-kafka-2.x-plugin pom diff --git a/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/pom.xml new file mode 100644 index 000000000..3e3461aed --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/pom.xml @@ -0,0 +1,50 @@ + + + + 4.0.0 + + + org.apache.skywalking + spring-plugins + 8.2.0-SNAPSHOT + + + apm-spring-kafka-2.x-plugin + + + 2.2.9.RELEASE + + + + + org.springframework.kafka + spring-kafka + ${spring-kafka.version} + provided + + + org.apache.skywalking + apm-kafka-commons + ${project.version} + provided + + + \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaTemplateCallbackInterceptor.java b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/KafkaTemplateCallbackInterceptor.java similarity index 84% rename from apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaTemplateCallbackInterceptor.java rename to apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/KafkaTemplateCallbackInterceptor.java index 99fea5572..d58d239e9 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaTemplateCallbackInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/KafkaTemplateCallbackInterceptor.java @@ -16,11 +16,12 @@ * */ -package org.apache.skywalking.apm.plugin.kafka; +package org.apache.skywalking.apm.plugin.spring.kafka; 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.kafka.CallbackAdapter; import java.lang.reflect.Method; @@ -30,19 +31,18 @@ import java.lang.reflect.Method; public class KafkaTemplateCallbackInterceptor implements InstanceMethodsAroundInterceptor { @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - MethodInterceptResult result) throws Throwable { + MethodInterceptResult result) throws Throwable { } @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, - Object ret) throws Throwable { + Object ret) throws Throwable { return new CallbackAdapter((org.apache.kafka.clients.producer.Callback) ret, objInst); } @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, - Class[] argumentsTypes, Throwable t) { - + Class[] argumentsTypes, Throwable t) { } } \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/PollAndInvokeMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/PollAndInvokeMethodInterceptor.java new file mode 100644 index 000000000..6a1056d4e --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/PollAndInvokeMethodInterceptor.java @@ -0,0 +1,57 @@ +/* + * 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.kafka; + +import org.apache.skywalking.apm.agent.core.context.ContextManager; +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.kafka.define.Constants; +import org.apache.skywalking.apm.plugin.kafka.define.SpringKafkaContext; + +import java.lang.reflect.Method; + +public class PollAndInvokeMethodInterceptor implements InstanceMethodsAroundInterceptor { + + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + ContextManager.getRuntimeContext().put(Constants.SPRING_KAFKA_FLAG, new SpringKafkaContext()); + } + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + SpringKafkaContext context = (SpringKafkaContext) ContextManager.getRuntimeContext().get(Constants.SPRING_KAFKA_FLAG); + if (context == null) { + return ret; + } + if (context.isNeedStop()) { + ContextManager.stopSpan(); + } else { + ContextManager.getRuntimeContext().remove(Constants.SPRING_KAFKA_FLAG); + } + return ret; + } + + @Override + public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaTemplateInstrumentation.java b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/define/KafkaTemplateInstrumentation.java similarity index 90% rename from apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaTemplateInstrumentation.java rename to apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/define/KafkaTemplateInstrumentation.java index b3752d33d..994c6f53f 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaTemplateInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/define/KafkaTemplateInstrumentation.java @@ -16,13 +16,14 @@ * */ -package org.apache.skywalking.apm.plugin.kafka.define; +package org.apache.skywalking.apm.plugin.spring.kafka.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.match.ClassMatch; +import org.apache.skywalking.apm.plugin.kafka.define.AbstractKafkaTemplateInstrumentation; import static net.bytebuddy.matcher.ElementMatchers.named; import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; @@ -34,7 +35,7 @@ public class KafkaTemplateInstrumentation extends AbstractKafkaTemplateInstrumen private static final String ENHANCE_CLASS = "org.springframework.kafka.core.KafkaTemplate"; private static final String ENHANCE_METHOD = "buildCallback"; - private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.KafkaTemplateCallbackInterceptor"; + private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.spring.kafka.KafkaTemplateCallbackInterceptor"; @Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { @@ -43,7 +44,7 @@ public class KafkaTemplateInstrumentation extends AbstractKafkaTemplateInstrumen @Override public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { - return new InstanceMethodsInterceptPoint[] { + return new InstanceMethodsInterceptPoint[]{ new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { diff --git a/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/define/ListenerConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/define/ListenerConsumerInstrumentation.java new file mode 100644 index 000000000..82c87543b --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/define/ListenerConsumerInstrumentation.java @@ -0,0 +1,65 @@ +/* + * 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.kafka.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 net.bytebuddy.matcher.ElementMatchers.takesArguments; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +public class ListenerConsumerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + @Override + public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[0]; + } + + @Override + public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[]{ + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named("pollAndInvoke").and(takesArguments(0)); + } + + @Override + public String getMethodsInterceptor() { + return "org.apache.skywalking.apm.plugin.spring.kafka.PollAndInvokeMethodInterceptor"; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + } + }; + } + + @Override + public ClassMatch enhanceClass() { + return byName("org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer"); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/resources/skywalking-plugin.def new file mode 100644 index 000000000..eeda8255a --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-plugin/src/main/resources/skywalking-plugin.def @@ -0,0 +1,18 @@ +# 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. + +spring-kafka-2.x=org.apache.skywalking.apm.plugin.spring.kafka.define.KafkaTemplateInstrumentation +spring-kafka-2.x=org.apache.skywalking.apm.plugin.spring.kafka.define.ListenerConsumerInstrumentation diff --git a/docs/en/FAQ/kafka-plugin.md b/docs/en/FAQ/kafka-plugin.md index d3d5efd58..2583f426f 100644 --- a/docs/en/FAQ/kafka-plugin.md +++ b/docs/en/FAQ/kafka-plugin.md @@ -5,4 +5,4 @@ The trace doesn't continue in kafka consumer side. The kafka client is pulling message from server, the plugin also just traces the pull action. As that, you need to do the manual instrument before the pull action, and include the further data process. ### Resolve -Use Application Toolkit libraries to do manual instrumentation. such as `@Trace` annotation or OpenTracing API. +Use Application Toolkit libraries to do manual instrumentation. such as `@Trace` annotation or OpenTracing API, Or if you're using `spring-kafka` 2.2.x or above, you can track the Consumer side without any code change. diff --git a/docs/en/setup/service-agent/java-agent/Supported-list.md b/docs/en/setup/service-agent/java-agent/Supported-list.md index 7c668e565..3d48a2ab1 100644 --- a/docs/en/setup/service-agent/java-agent/Supported-list.md +++ b/docs/en/setup/service-agent/java-agent/Supported-list.md @@ -49,6 +49,7 @@ * MQ * [RocketMQ](https://github.com/apache/rocketmq) 4.x * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 1.0 + * [Spring-Kafka](https://github.com/spring-projects/spring-kafka) Spring Kafka Consumer 2.2.x * [ActiveMQ](https://github.com/apache/activemq) 5.10.0 -> 5.15.4 * [RabbitMQ](https://www.rabbitmq.com/) 5.x * [Pulsar](http://pulsar.apache.org) 2.2.x -> 2.4.x diff --git a/oap-server/server-bootstrap/src/main/resources/component-libraries.yml b/oap-server/server-bootstrap/src/main/resources/component-libraries.yml index 36e2d47c2..7c464a757 100755 --- a/oap-server/server-bootstrap/src/main/resources/component-libraries.yml +++ b/oap-server/server-bootstrap/src/main/resources/component-libraries.yml @@ -314,6 +314,9 @@ spring-annotation: HBase: id: 94 languages: Java +spring-kafka-consumer: + id: 95 + languages: Java # .NET/.NET Core components # [3000, 4000) for C#/.NET only @@ -496,4 +499,5 @@ Component-Server-Mappings: Mysqli: Mysql influxdb-java: InfluxDB Predis: Redis - PyMysql: Mysql \ No newline at end of file + PyMysql: Mysql + spring-kafka-consumer: kafka-consumer \ No newline at end of file diff --git a/test/plugin/agent-test-tools/pom.xml b/test/plugin/agent-test-tools/pom.xml index 67009761e..db3754c75 100644 --- a/test/plugin/agent-test-tools/pom.xml +++ b/test/plugin/agent-test-tools/pom.xml @@ -35,7 +35,7 @@ pom - 08f6fc9466d892e25794c4a9561750816bd06d28 + 2eb5cb96c36fd720a013c70dba8c5717ddfbe870 ${project.basedir}/target/agent-test-tools https://github.com/apache/skywalking-agent-test-tool.git diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/bin/startup.sh b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/bin/startup.sh new file mode 100644 index 000000000..fafb421be --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/bin/startup.sh @@ -0,0 +1,21 @@ +#!/bin/bash +# +# 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. + +home="$(cd "$(dirname $0)"; pwd)" + +java -Dbootstrap.servers=${BOOTSTRAP_SERVERS} -jar ${agent_opts} "-Dskywalking.agent.service_name=spring-kafka-2.2.x-scenario" ${home}/../libs/spring-kafka-2.2.x-scenario.jar & \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/config/expectedData.yaml new file mode 100644 index 000000000..46f495ce9 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/config/expectedData.yaml @@ -0,0 +1,110 @@ +# 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. +segmentItems: + - serviceName: spring-kafka-2.2.x-scenario + segmentSize: nq 0 + segments: + - segmentId: not null + spans: + - operationName: Kafka/spring_test/Producer + operationId: 0 + parentSpanId: 0 + spanId: 1 + spanLayer: MQ + startTime: not null + endTime: not null + componentId: 40 + isError: false + spanType: Exit + peer: kafka-server:9092 + skipAnalysis: false + tags: + - {key: mq.broker, value: 'kafka-server:9092'} + - {key: mq.topic, value: spring_test} + - operationName: /case/spring-kafka-case + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 14 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: url, value: 'http://localhost:8080/spring-kafka-2.2.x-scenario/case/spring-kafka-case'} + - {key: http.method, value: GET} + - segmentId: not null + spans: + - operationName: /case/spring-kafka-consumer-ping + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 14 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: url, value: 'http://localhost:8080/spring-kafka-2.2.x-scenario/case/spring-kafka-consumer-ping'} + - {key: http.method, value: GET} + refs: + - {parentEndpoint: 'Kafka/spring_test/Consumer/grop:spring_test', networkAddress: 'localhost:8080', + refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: spring-kafka-2.2.x-scenario, + traceId: not null} + - segmentId: not null + spans: + - operationName: /spring-kafka-2.2.x-scenario/case/spring-kafka-consumer-ping + operationId: 0 + parentSpanId: 0 + spanId: 1 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 12 + isError: false + spanType: Exit + peer: localhost:8080 + skipAnalysis: false + tags: + - {key: http.method, value: GET} + - {key: url, value: 'http://localhost:8080/spring-kafka-2.2.x-scenario/case/spring-kafka-consumer-ping'} + - operationName: Kafka/spring_test/Consumer/grop:spring_test + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: MQ + startTime: not null + endTime: not null + componentId: 41 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: mq.broker, value: 'kafka-server:9092'} + - {key: mq.topic, value: spring_test} + refs: + - {parentEndpoint: /case/spring-kafka-case, networkAddress: 'kafka-server:9092', + refType: CrossProcess, parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: spring-kafka-2.2.x-scenario, + traceId: not null} diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/configuration.yml b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/configuration.yml new file mode 100644 index 000000000..c8b94d2be --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/configuration.yml @@ -0,0 +1,39 @@ +# 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. + +type: jvm +entryService: http://localhost:8080/spring-kafka-2.2.x-scenario/case/spring-kafka-case +healthCheck: http://localhost:8080/spring-kafka-2.2.x-scenario/case/healthCheck +startScript: ./bin/startup.sh +environment: + - BOOTSTRAP_SERVERS=kafka-server:9092 +depends_on: + - zookeeper-server + - kafka-server +dependencies: + zookeeper-server: + image: zookeeper:3.4 + hostname: zookeeper-server + kafka-server: + image: bitnami/kafka:2.1.1 + hostname: kafka-server + environment: + - KAFKA_ZOOKEEPER_CONNECT=zookeeper-server:2181 + - KAFKA_BROKER_ID=1 + - ALLOW_PLAINTEXT_LISTENER=yes + - KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 + depends_on: + - zookeeper-server diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/pom.xml b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/pom.xml new file mode 100644 index 000000000..0c0d99111 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/pom.xml @@ -0,0 +1,134 @@ + + + + 4.0.0 + + org.apache.skywalking + spring-kafka-2.2.x-scenario + 5.0.0 + + + UTF-8 + 1.8 + 2.2.9.RELEASE + 2.6.2 + 2.1.9.RELEASE + 2.0.1 + 3.0.0 + + + skywalking-spring-kafka-2.2.x-scenario + + + + org.springframework.boot + spring-boot-starter-web + ${spring-boot-version} + + + org.springframework.boot + spring-boot-starter-jdbc + ${spring-boot-version} + + + org.springframework.kafka + spring-kafka + ${test.framework.version} + + + org.slf4j + * + + + + + org.apache.kafka + kafka-clients + ${kafka-version} + + + slf4j-api + * + + + + + com.squareup.okhttp3 + okhttp + ${okhttp-version} + + + + + spring-kafka-2.2.x-scenario + + + org.springframework.boot + spring-boot-maven-plugin + + + + repackage + + + + + + maven-compiler-plugin + + ${compiler.version} + ${compiler.version} + ${project.build.sourceEncoding} + + + + org.apache.maven.plugins + maven-assembly-plugin + + + assemble + package + + single + + + + src/main/assembly/assembly.xml + + ./target/ + + + + + + + + + + spring-snapshots + http://repo.spring.io/snapshot + + + spring-milestones + http://repo.spring.io/milestone + + + \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/assembly/assembly.xml b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/assembly/assembly.xml new file mode 100644 index 000000000..14d8f7aa5 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/assembly/assembly.xml @@ -0,0 +1,41 @@ + + + + + zip + + + + + ./bin + 0775 + + + + + + ${project.build.directory}/spring-kafka-2.2.x-scenario.jar + ./libs + 0775 + + + diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/Application.java b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/Application.java new file mode 100644 index 000000000..b35127e7c --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/Application.java @@ -0,0 +1,32 @@ +/* + * 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 test.org.apache.skywalking.apm.testcase.spring.kafka; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; + +@SpringBootApplication(exclude={DataSourceAutoConfiguration.class}) +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } +} diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java new file mode 100644 index 000000000..e5dc9f086 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java @@ -0,0 +1,127 @@ +/* + * 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 test.org.apache.skywalking.apm.testcase.spring.kafka.controller; + +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.PropertySource; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.AcknowledgingMessageListener; +import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.ResponseBody; + +import javax.annotation.PostConstruct; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +@Controller +@RequestMapping("/case") +@PropertySource("classpath:application.properties") +public class CaseController { + + private static final String SUCCESS = "Success"; + + @Value("${bootstrap.servers:127.0.0.1:9092}") + private String bootstrapServers; + private String topicName; + private KafkaTemplate kafkaTemplate; + + @PostConstruct + private void setUp() { + topicName = "spring_test"; + setUpProvider(); + setUpConsumer(); + } + + private void setUpProvider() { + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + kafkaTemplate = new KafkaTemplate(new DefaultKafkaProducerFactory<>(props)); + } + + private void setUpConsumer() { + Map configs = new HashMap<>(); + configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + configs.put(ConsumerConfig.GROUP_ID_CONFIG, "grop:" + topicName); + configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + Deserializer stringDeserializer = new StringDeserializer(); + DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(configs, stringDeserializer, stringDeserializer); + ContainerProperties props = new ContainerProperties(topicName); + props.setMessageListener(new AcknowledgingMessageListener() { + @Override + public void onMessage(ConsumerRecord data, Acknowledgment acknowledgment) { + System.out.println(data); + OkHttpClient client = new OkHttpClient.Builder().build(); + Request request = new Request.Builder().url("http://localhost:8080/spring-kafka-2.2.x-scenario/case/spring-kafka-consumer-ping").build(); + Response response = null; + try { + response = client.newCall(request).execute(); + } catch (IOException e) { + } + response.body().close(); + acknowledgment.acknowledge(); + } + }); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(factory, props); + container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); + container.start(); + } + + @RequestMapping("/spring-kafka-case") + @ResponseBody + public String springKafkaCase() { + try { + kafkaTemplate.send(topicName, "key", "helloWorld").get(); + Thread.sleep(2000L); + } catch (Exception e) { + e.printStackTrace(); + } + return SUCCESS; + } + + @RequestMapping("/spring-kafka-consumer-ping") + @ResponseBody + public String springKafkaConsumerPing() { + return SUCCESS; + } + + @RequestMapping("/healthCheck") + @ResponseBody + public String healthCheck() { + return SUCCESS; + } +} + diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/resources/application.properties b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/resources/application.properties new file mode 100644 index 000000000..9e230a95a --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/resources/application.properties @@ -0,0 +1,19 @@ +# +# 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. +# +# +server.port=8080 +server.servlet.context-path=/spring-kafka-2.2.x-scenario \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/resources/log4j2.xml b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/resources/log4j2.xml new file mode 100644 index 000000000..b5cda5ae8 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/resources/log4j2.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/support-version.list b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/support-version.list new file mode 100644 index 000000000..d576444f6 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/support-version.list @@ -0,0 +1,23 @@ +# 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 +# "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. + +2.2.14.RELEASE +2.2.13.RELEASE +2.2.12.RELEASE +2.2.11.RELEASE +2.2.10.RELEASE +2.2.9.RELEASE +2.2.8.RELEASE \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/bin/startup.sh b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/bin/startup.sh new file mode 100644 index 000000000..20c7ee200 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/bin/startup.sh @@ -0,0 +1,21 @@ +#!/bin/bash +# +# 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. + +home="$(cd "$(dirname $0)"; pwd)" + +java -Dbootstrap.servers=${BOOTSTRAP_SERVERS} -jar ${agent_opts} "-Dskywalking.agent.service_name=spring-kafka-2.3.x-scenario" ${home}/../libs/spring-kafka-2.3.x-scenario.jar & \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml new file mode 100644 index 000000000..74116c0df --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml @@ -0,0 +1,110 @@ +# 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. +segmentItems: + - serviceName: spring-kafka-2.3.x-scenario + segmentSize: nq 0 + segments: + - segmentId: not null + spans: + - operationName: Kafka/spring_test/Producer + operationId: 0 + parentSpanId: 0 + spanId: 1 + spanLayer: MQ + startTime: not null + endTime: not null + componentId: 40 + isError: false + spanType: Exit + peer: kafka-server:9092 + skipAnalysis: false + tags: + - {key: mq.broker, value: 'kafka-server:9092'} + - {key: mq.topic, value: spring_test} + - operationName: /case/spring-kafka-case + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 14 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: url, value: 'http://localhost:8080/spring-kafka-2.3.x-scenario/case/spring-kafka-case'} + - {key: http.method, value: GET} + - segmentId: not null + spans: + - operationName: /case/spring-kafka-consumer-ping + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 14 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: url, value: 'http://localhost:8080/spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping'} + - {key: http.method, value: GET} + refs: + - {parentEndpoint: 'Kafka/spring_test/Consumer/grop:spring_test', networkAddress: 'localhost:8080', + refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: spring-kafka-2.3.x-scenario, + traceId: not null} + - segmentId: not null + spans: + - operationName: /spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping + operationId: 0 + parentSpanId: 0 + spanId: 1 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 12 + isError: false + spanType: Exit + peer: localhost:8080 + skipAnalysis: false + tags: + - {key: http.method, value: GET} + - {key: url, value: 'http://localhost:8080/spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping'} + - operationName: Kafka/spring_test/Consumer/grop:spring_test + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: MQ + startTime: not null + endTime: not null + componentId: 41 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: mq.broker, value: 'kafka-server:9092'} + - {key: mq.topic, value: spring_test} + refs: + - {parentEndpoint: /case/spring-kafka-case, networkAddress: 'kafka-server:9092', + refType: CrossProcess, parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: spring-kafka-2.3.x-scenario, + traceId: not null} diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/configuration.yml b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/configuration.yml new file mode 100644 index 000000000..3d40a5235 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/configuration.yml @@ -0,0 +1,39 @@ +# 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. + +type: jvm +entryService: http://localhost:8080/spring-kafka-2.3.x-scenario/case/spring-kafka-case +healthCheck: http://localhost:8080/spring-kafka-2.3.x-scenario/case/healthCheck +startScript: ./bin/startup.sh +environment: + - BOOTSTRAP_SERVERS=kafka-server:9092 +depends_on: + - zookeeper-server + - kafka-server +dependencies: + zookeeper-server: + image: zookeeper:3.4 + hostname: zookeeper-server + kafka-server: + image: bitnami/kafka:2.1.1 + hostname: kafka-server + environment: + - KAFKA_ZOOKEEPER_CONNECT=zookeeper-server:2181 + - KAFKA_BROKER_ID=1 + - ALLOW_PLAINTEXT_LISTENER=yes + - KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 + depends_on: + - zookeeper-server diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/pom.xml b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/pom.xml new file mode 100644 index 000000000..948b14c68 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/pom.xml @@ -0,0 +1,129 @@ + + + + 4.0.0 + + org.apache.skywalking + spring-kafka-2.3.x-scenario + 5.0.0 + + + UTF-8 + 1.8 + 2.3.3.RELEASE + 2.6.2 + 2.3.2.RELEASE + 2.3.1 + 3.0.0 + + + skywalking-spring-kafka-2.3.x-scenario + + + + org.springframework.boot + spring-boot-starter-web + ${spring-boot-version} + + + org.springframework.kafka + spring-kafka + ${test.framework.version} + + + org.slf4j + * + + + + + org.apache.kafka + kafka-clients + ${kafka-version} + + + slf4j-api + * + + + + + com.squareup.okhttp3 + okhttp + ${okhttp-version} + + + + + spring-kafka-2.3.x-scenario + + + org.springframework.boot + spring-boot-maven-plugin + + + + repackage + + + + + + maven-compiler-plugin + + ${compiler.version} + ${compiler.version} + ${project.build.sourceEncoding} + + + + org.apache.maven.plugins + maven-assembly-plugin + + + assemble + package + + single + + + + src/main/assembly/assembly.xml + + ./target/ + + + + + + + + + + spring-snapshots + http://repo.spring.io/snapshot + + + spring-milestones + http://repo.spring.io/milestone + + + \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/assembly/assembly.xml b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/assembly/assembly.xml new file mode 100644 index 000000000..e4ba8c3fc --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/assembly/assembly.xml @@ -0,0 +1,41 @@ + + + + + zip + + + + + ./bin + 0775 + + + + + + ${project.build.directory}/spring-kafka-2.3.x-scenario.jar + ./libs + 0775 + + + diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/Application.java b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/Application.java new file mode 100644 index 000000000..afefaeda9 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/Application.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 test.org.apache.skywalking.apm.testcase.spring.kafka; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } +} diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java new file mode 100644 index 000000000..5b3f98722 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java @@ -0,0 +1,127 @@ +/* + * 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 test.org.apache.skywalking.apm.testcase.spring.kafka.controller; + +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.PropertySource; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.AcknowledgingMessageListener; +import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.ResponseBody; + +import javax.annotation.PostConstruct; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +@Controller +@RequestMapping("/case") +@PropertySource("classpath:application.properties") +public class CaseController { + + private static final String SUCCESS = "Success"; + + @Value("${bootstrap.servers:127.0.0.1:9092}") + private String bootstrapServers; + private String topicName; + private KafkaTemplate kafkaTemplate; + + @PostConstruct + private void setUp() { + topicName = "spring_test"; + setUpProvider(); + setUpConsumer(); + } + + private void setUpProvider() { + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + kafkaTemplate = new KafkaTemplate(new DefaultKafkaProducerFactory<>(props)); + } + + private void setUpConsumer() { + Map configs = new HashMap<>(); + configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + configs.put(ConsumerConfig.GROUP_ID_CONFIG, "grop:" + topicName); + configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + Deserializer stringDeserializer = new StringDeserializer(); + DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(configs, stringDeserializer, stringDeserializer); + ContainerProperties props = new ContainerProperties(topicName); + props.setMessageListener(new AcknowledgingMessageListener() { + @Override + public void onMessage(ConsumerRecord data, Acknowledgment acknowledgment) { + System.out.println(data); + OkHttpClient client = new OkHttpClient.Builder().build(); + Request request = new Request.Builder().url("http://localhost:8080/spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping").build(); + Response response = null; + try { + response = client.newCall(request).execute(); + } catch (IOException e) { + } + response.body().close(); + acknowledgment.acknowledge(); + } + }); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(factory, props); + container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); + container.start(); + } + + @RequestMapping("/spring-kafka-case") + @ResponseBody + public String springKafkaCase() { + try { + kafkaTemplate.send(topicName, "key", "helloWorld").get(); + Thread.sleep(2000L); + } catch (Exception e) { + e.printStackTrace(); + } + return SUCCESS; + } + + @RequestMapping("/spring-kafka-consumer-ping") + @ResponseBody + public String springKafkaConsumerPing() { + return SUCCESS; + } + + @RequestMapping("/healthCheck") + @ResponseBody + public String healthCheck() { + return SUCCESS; + } +} + diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/resources/application.properties b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/resources/application.properties new file mode 100644 index 000000000..2db88eca5 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/resources/application.properties @@ -0,0 +1,19 @@ +# +# 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. +# +# +server.port=8080 +server.servlet.context-path=/spring-kafka-2.3.x-scenario \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/resources/log4j2.xml b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/resources/log4j2.xml new file mode 100644 index 000000000..b5cda5ae8 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/resources/log4j2.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/support-version.list b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/support-version.list new file mode 100644 index 000000000..8fc1bada8 --- /dev/null +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/support-version.list @@ -0,0 +1,23 @@ +# 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 +# "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. + +2.3.10.RELEASE +2.3.9.RELEASE +2.3.8.RELEASE +2.3.7.RELEASE +2.3.6.RELEASE +2.3.5.RELEASE +2.3.4.RELEASE \ No newline at end of file