From 906a2c4d05755112a94e494b68a0de4b716f3dea Mon Sep 17 00:00:00 2001 From: ascrutae Date: Tue, 23 Jan 2018 20:03:50 +0800 Subject: [PATCH] add testcase --- .../core/context/trace/AbstractSpan.java | 2 + .../context/trace/AbstractTracingSpan.java | 6 + .../agent/core/context/trace/NoopSpan.java | 4 + .../kafka-0.11.x-plugin/README.md | 5 + .../v11/ConsumerEnhanceRequiredInfo.java | 13 +- .../kafka/v11/KafkaConsumerInterceptor.java | 47 +++--- .../kafka/v11/KafkaProducerInterceptor.java | 13 +- .../v11/ProducerConstructorInterceptor.java | 2 +- .../kafka/v11/CallbackInterceptorTest.java | 147 ++++++++++++++++++ .../ConsumerConstructorInterceptorTest.java | 73 +++++++++ .../v11/KafkaConsumerInterceptorTest.java | 140 +++++++++++++++++ .../v11/KafkaProducerInterceptorTest.java | 115 ++++++++++++++ .../ProducerConstructorInterceptorTest.java | 69 ++++++++ ...ducerRecordConstructorInterceptorTest.java | 59 +++++++ .../v11/SubscribeMethodInterceptorTest.java | 67 ++++++++ .../core/context/MockContextSnapshot.java | 47 ++++++ 16 files changed, 785 insertions(+), 24 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.java create mode 100644 apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractSpan.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractSpan.java index ada06d3d0..3ce6518d3 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractSpan.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractSpan.java @@ -116,4 +116,6 @@ public interface AbstractSpan { * @param ref segment ref */ void ref(TraceSegmentRef ref); + + AbstractSpan start(long starttime); } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java index 90d174d02..2028215ae 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java @@ -239,6 +239,12 @@ public abstract class AbstractTracingSpan implements AbstractSpan { return this; } + @Override + public AbstractSpan start(long startTime) { + this.startTime = startTime; + return this; + } + public SpanObject.Builder transform() { SpanObject.Builder spanBuilder = SpanObject.newBuilder(); diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/NoopSpan.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/NoopSpan.java index 4be30d1b7..df64b9015 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/NoopSpan.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/NoopSpan.java @@ -102,4 +102,8 @@ public class NoopSpan implements AbstractSpan { @Override public void ref(TraceSegmentRef ref) { } + + @Override public AbstractSpan start(long startTime) { + return this; + } } diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md new file mode 100644 index 000000000..bcc6c3b46 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md @@ -0,0 +1,5 @@ +# Kafka Plugin + +## +## Buired Point + diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java index 2eb660137..97e44c57a 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java @@ -25,13 +25,14 @@ import org.apache.skywalking.apm.util.StringUtil; public class ConsumerEnhanceRequiredInfo { private String brokerServers; private String topics; + private long startTime; public void setBrokerServers(List brokerServers) { - this.brokerServers = StringUtil.join(',', brokerServers.toArray(new String[0])); + this.brokerServers = StringUtil.join(';', brokerServers.toArray(new String[0])); } public void setTopics(Collection topics) { - this.topics = StringUtil.join(',', topics.toArray(new String[0])); + this.topics = StringUtil.join(';', topics.toArray(new String[0])); } public String getBrokerServers() { @@ -41,4 +42,12 @@ public class ConsumerEnhanceRequiredInfo { public String getTopics() { return topics; } + + public void setStartTime(long startTime) { + this.startTime = startTime; + } + + public long getStartTime() { + return startTime; + } } diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java index f6804cc2f..f0098e43d 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java @@ -40,39 +40,50 @@ import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; * @autor zhang xin */ public class KafkaConsumerInterceptor implements InstanceMethodsAroundInterceptor { + + public static final String OPERATE_NAME_PREFIX = "Kafka/"; + public static final String CONSUMER_OPERATE_NAME_SUFFIX = "/Consumer"; + @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, MethodInterceptResult result) throws Throwable { - AbstractSpan activeSpan = ContextManager.createEntrySpan("Kafka/Consumer/Poll", null); ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo)objInst.getSkyWalkingDynamicField(); - - activeSpan.setComponent(ComponentsDefine.KAFKA); - SpanLayer.asMQ(activeSpan); - Tags.MQ_BROKER.set(activeSpan, requiredInfo.getBrokerServers()); - Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopics()); + requiredInfo.setStartTime(System.currentTimeMillis()); } @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { Map>> records = (Map>>)ret; - for (List> consumerRecords : records.values()) { - for (ConsumerRecord record : consumerRecords) { - ContextCarrier contextCarrier = new ContextCarrier(); + // + // The entry span will create when the consumer fetch anyone message from kafka cluster, or the span will not create. + // + if (records.size() > 0) { + ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo)objInst.getSkyWalkingDynamicField(); + AbstractSpan activeSpan = ContextManager.createEntrySpan(OPERATE_NAME_PREFIX + requiredInfo.getTopics() + CONSUMER_OPERATE_NAME_SUFFIX, null).start(requiredInfo.getStartTime()); - CarrierItem next = contextCarrier.items(); - while (next.hasNext()) { - next = next.next(); - Iterator
iterator = record.headers().headers(next.getHeadKey()).iterator(); - if (iterator.hasNext()) { - next.setHeadValue(new String(iterator.next().value())); + activeSpan.setComponent(ComponentsDefine.KAFKA); + SpanLayer.asMQ(activeSpan); + Tags.MQ_BROKER.set(activeSpan, requiredInfo.getBrokerServers()); + Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopics()); + + for (List> consumerRecords : records.values()) { + for (ConsumerRecord record : consumerRecords) { + ContextCarrier contextCarrier = new ContextCarrier(); + + CarrierItem next = contextCarrier.items(); + while (next.hasNext()) { + next = next.next(); + Iterator
iterator = record.headers().headers(next.getHeadKey()).iterator(); + if (iterator.hasNext()) { + next.setHeadValue(new String(iterator.next().value())); + } } + ContextManager.extract(contextCarrier); } - ContextManager.extract(contextCarrier); } + ContextManager.stopSpan(); } - - ContextManager.stopSpan(); return ret; } diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java index 70d98744b..19d883b70 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java @@ -35,17 +35,24 @@ import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; * @author zhang xin */ public class KafkaProducerInterceptor implements InstanceMethodsAroundInterceptor { + + public static final String OPERATE_NAME_PREFIX = "Kafka/"; + public static final String PRODUCER_OPERATE_NAME_SUFFIX = "/Producer"; + @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, MethodInterceptResult result) throws Throwable { ContextCarrier contextCarrier = new ContextCarrier(); - AbstractSpan activeSpan = ContextManager.createExitSpan("Kafka/Producer/send", contextCarrier, (String)objInst.getSkyWalkingDynamicField()); + + ProducerRecord record = (ProducerRecord)allArguments[0]; + String topicName = (String)((EnhancedInstance)record).getSkyWalkingDynamicField(); + + AbstractSpan activeSpan = ContextManager.createExitSpan(OPERATE_NAME_PREFIX + topicName + PRODUCER_OPERATE_NAME_SUFFIX, contextCarrier, (String)objInst.getSkyWalkingDynamicField()); //set tags - ProducerRecord record = (ProducerRecord)allArguments[0]; Tags.MQ_BROKER.set(activeSpan, (String)objInst.getSkyWalkingDynamicField()); - Tags.MQ_TOPIC.set(activeSpan, (String)((EnhancedInstance)record).getSkyWalkingDynamicField()); + Tags.MQ_TOPIC.set(activeSpan, topicName); SpanLayer.asMQ(activeSpan); activeSpan.setComponent(ComponentsDefine.KAFKA); diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java index ed267df21..9849d48ff 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java @@ -28,6 +28,6 @@ public class ProducerConstructorInterceptor implements InstanceConstructorInterc @Override public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { ProducerConfig config = (ProducerConfig)allArguments[0]; // set the bootstrap server address - objInst.setSkyWalkingDynamicField(StringUtil.join(',', config.getList("bootstrap.servers").toArray(new String[0]))); + objInst.setSkyWalkingDynamicField(StringUtil.join(';', config.getList("bootstrap.servers").toArray(new String[0]))); } } diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java new file mode 100644 index 000000000..78991cfe2 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java @@ -0,0 +1,147 @@ +/* + * 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.v11; + +import java.util.List; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.skywalking.apm.agent.core.context.MockContextSnapshot; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegmentRef; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.test.helper.SegmentHelper; +import org.apache.skywalking.apm.agent.test.helper.SpanHelper; +import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule; +import org.apache.skywalking.apm.agent.test.tools.SegmentRefAssert; +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.SpanAssert; +import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; +import org.powermock.modules.junit4.PowerMockRunnerDelegate; + +import static org.hamcrest.core.Is.is; +import static org.junit.Assert.assertThat; + +@RunWith(PowerMockRunner.class) +@PowerMockRunnerDelegate(TracingSegmentRunner.class) +@PrepareForTest({RecordMetadata.class}) +public class CallbackInterceptorTest { + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + + @Mock + private RecordMetadata recordMetadata; + + private CallbackInterceptor callbackInterceptor; + + private Object[] arguments; + private Object[] argumentsWithException; + private Class[] argumentTypes; + + private EnhancedInstance callBackInstance = new EnhancedInstance() { + @Override public Object getSkyWalkingDynamicField() { + return MockContextSnapshot.INSTANCE.mockContextSnapshot(); + } + + @Override public void setSkyWalkingDynamicField(Object value) { + + } + }; + + @Before + public void setUp() { + callbackInterceptor = new CallbackInterceptor(); + + arguments = new Object[] { + recordMetadata, null + }; + argumentsWithException = new Object[] { + recordMetadata, new RuntimeException() + }; + + argumentTypes = new Class[] { + RecordMetadata.class, Exception.class + }; + + } + + @Test + public void testCallbackWithoutException() throws Throwable { + callbackInterceptor.beforeMethod(callBackInstance, null, arguments, argumentTypes, null); + callbackInterceptor.afterMethod(callBackInstance, null, arguments, argumentTypes, null); + + List traceSegments = segmentStorage.getTraceSegments(); + assertThat(traceSegments.size(), is(1)); + TraceSegment traceSegment = traceSegments.get(0); + + List abstractSpans = SegmentHelper.getSpans(traceSegment); + assertThat(abstractSpans.size(), is(1)); + + assertCallbackSpan(abstractSpans.get(0)); + + assertCallbackSegmentRef(traceSegment.getRefs()); + } + + @Test + public void testCallbackWithException() throws Throwable { + callbackInterceptor.beforeMethod(callBackInstance, null, argumentsWithException, argumentTypes, null); + callbackInterceptor.afterMethod(callBackInstance, null, argumentsWithException, argumentTypes, null); + + List traceSegments = segmentStorage.getTraceSegments(); + assertThat(traceSegments.size(), is(1)); + TraceSegment traceSegment = traceSegments.get(0); + + List abstractSpans = SegmentHelper.getSpans(traceSegment); + assertThat(abstractSpans.size(), is(1)); + + assertCallbackSpanWithException(abstractSpans.get(0)); + + assertCallbackSegmentRef(traceSegment.getRefs()); + } + + private void assertCallbackSpanWithException(AbstractTracingSpan span) { + assertCallbackSpan(span); + + SpanAssert.assertException(SpanHelper.getLogs(span).get(0), RuntimeException.class); + assertThat(SpanHelper.getErrorOccurred(span), is(true)); + } + + private void assertCallbackSegmentRef(List refs) { + assertThat(refs.size(), is(1)); + + TraceSegmentRef segmentRef = refs.get(0); + SegmentRefAssert.assertSpanId(segmentRef, 1); + assertThat(segmentRef.getEntryOperationName(), is("/for-test-entryOperationName")); + } + + private void assertCallbackSpan(AbstractTracingSpan span) { + assertThat(span.getOperationName(), is("Producer/Callback")); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java new file mode 100644 index 000000000..1379203d6 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java @@ -0,0 +1,73 @@ +/* + * 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.v11; + +import java.util.ArrayList; +import java.util.List; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class ConsumerConstructorInterceptorTest { + + @Mock + private ConsumerConfig consumerConfig; + + @Mock + private ConsumerConstructorInterceptor constructorInterceptor; + + private EnhancedInstance enhancedInstance = new EnhancedInstance() { + private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo; + + @Override public Object getSkyWalkingDynamicField() { + return consumerEnhanceRequiredInfo; + } + + @Override public void setSkyWalkingDynamicField(Object value) { + consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)value; + } + }; + + @Before + public void setUp() { + List mockBootstrapServers = new ArrayList(); + mockBootstrapServers.add("localhost:9092"); + mockBootstrapServers.add("localhost:19092"); + when(consumerConfig.getList("bootstrap.servers")).thenReturn(mockBootstrapServers); + + constructorInterceptor = new ConsumerConstructorInterceptor(); + } + + @Test + public void testOnConsumer() { + constructorInterceptor.onConstruct(enhancedInstance, new Object[] {consumerConfig}); + ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)enhancedInstance.getSkyWalkingDynamicField(); + assertThat(consumerEnhanceRequiredInfo.getBrokerServers(), is("localhost:9092;localhost:19092")); + } + +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java new file mode 100644 index 000000000..847b9ea87 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java @@ -0,0 +1,140 @@ +/* + * 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.v11; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegmentRef; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.test.helper.SegmentHelper; +import org.apache.skywalking.apm.agent.test.helper.SegmentRefHelper; +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.SpanAssert; +import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.hamcrest.MatcherAssert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.modules.junit4.PowerMockRunner; +import org.powermock.modules.junit4.PowerMockRunnerDelegate; + +import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.KAFKA; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; + +@RunWith(PowerMockRunner.class) +@PowerMockRunnerDelegate(TracingSegmentRunner.class) +public class KafkaConsumerInterceptorTest { + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + + private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo; + + private KafkaConsumerInterceptor consumerInterceptor; + + private EnhancedInstance consumerInstance = new EnhancedInstance() { + @Override public Object getSkyWalkingDynamicField() { + return consumerEnhanceRequiredInfo; + } + + @Override public void setSkyWalkingDynamicField(Object value) { + consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)value; + } + }; + + private Map> messages; + + @Before + public void setUp() { + consumerInterceptor = new KafkaConsumerInterceptor(); + consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo(); + + List topics = new ArrayList(); + topics.add("test"); + topics.add("test-1"); + consumerEnhanceRequiredInfo.setTopics(topics); + List brokers = new ArrayList(); + brokers.add("localhost:9092"); + brokers.add("localhost:19092"); + consumerEnhanceRequiredInfo.setBrokerServers(brokers); + + messages = new HashMap>(); + TopicPartition topicPartition = new TopicPartition("test", 1); + List records = new ArrayList(); + ConsumerRecord consumerRecord = new ConsumerRecord("test", 1, 0, "1", "1"); + consumerRecord.headers().add("sw3", "1.234.111|3|1|1|#192.168.1.8:18002|#/portal/|#testEntrySpan|#AQA*#AQA*Et0We0tQNQA*".getBytes()); + records.add(consumerRecord); + messages.put(topicPartition, records); + } + + @Test + public void testConsumerWithoutMessage() throws Throwable { + consumerInterceptor.beforeMethod(consumerInstance, null, new Object[0], new Class[0], null); + consumerInterceptor.afterMethod(consumerInstance, null, new Object[0], new Class[0], new HashMap>()); + + List traceSegments = segmentStorage.getTraceSegments(); + assertThat(traceSegments.size(), is(0)); + } + + @Test + public void testConsumerWithMessage() throws Throwable { + consumerInterceptor.beforeMethod(consumerInstance, null, new Object[0], new Class[0], null); + consumerInterceptor.afterMethod(consumerInstance, null, new Object[0], new Class[0], messages); + + List traceSegments = segmentStorage.getTraceSegments(); + assertThat(traceSegments.size(), is(1)); + + TraceSegment traceSegment = traceSegments.get(0); + List refs = traceSegment.getRefs(); + assertThat(refs.size(), is(1)); + assertTraceSegmentRef(refs.get(0)); + + List spans = SegmentHelper.getSpans(traceSegment); + assertThat(spans.size(), is(1)); + assertConsumerSpan(spans.get(0)); + } + + private void assertConsumerSpan(AbstractTracingSpan span) { + SpanAssert.assertLayer(span, SpanLayer.MQ); + SpanAssert.assertComponent(span, KAFKA); + SpanAssert.assertTagSize(span, 2); + SpanAssert.assertTag(span, 0, "localhost:9092;localhost:19092"); + SpanAssert.assertTag(span, 1, "test;test-1"); + } + + private void assertTraceSegmentRef(TraceSegmentRef ref) { + MatcherAssert.assertThat(SegmentRefHelper.getEntryApplicationInstanceId(ref), is(1)); + MatcherAssert.assertThat(SegmentRefHelper.getSpanId(ref), is(3)); + MatcherAssert.assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("1.234.111")); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java new file mode 100644 index 000000000..3c7644112 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java @@ -0,0 +1,115 @@ +/* + * 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.v11; + +import java.util.List; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.test.helper.SegmentHelper; +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.SpanAssert; +import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.modules.junit4.PowerMockRunner; +import org.powermock.modules.junit4.PowerMockRunnerDelegate; + +import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.KAFKA; +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; + +@RunWith(PowerMockRunner.class) +@PowerMockRunnerDelegate(TracingSegmentRunner.class) +public class KafkaProducerInterceptorTest { + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + + private KafkaProducerInterceptor producerInterceptor; + + private Object[] arguments; + private Class[] argumentType; + + private EnhancedInstance kafkaProducerInstance = new EnhancedInstance() { + @Override public Object getSkyWalkingDynamicField() { + return "localhost:9092"; + } + + @Override public void setSkyWalkingDynamicField(Object value) { + + } + }; + + private EnhancedInstance messageInstance = new MockProducerMessage(); + + private class MockProducerMessage extends ProducerRecord implements EnhancedInstance { + + public MockProducerMessage() { + super("test", ""); + } + + @Override public Object getSkyWalkingDynamicField() { + return "test"; + } + + @Override public void setSkyWalkingDynamicField(Object value) { + + } + } + + @Before + public void setUp() { + producerInterceptor = new KafkaProducerInterceptor(); + + arguments = new Object[] {messageInstance, null}; + argumentType = new Class[] {ProducerRecord.class}; + } + + @Test + public void testSendMessage() throws Throwable { + producerInterceptor.beforeMethod(kafkaProducerInstance, null, arguments, argumentType, null); + producerInterceptor.afterMethod(kafkaProducerInstance, null, arguments, argumentType, null); + + List traceSegmentList = segmentStorage.getTraceSegments(); + assertThat(traceSegmentList.size(), is(1)); + + TraceSegment segment = traceSegmentList.get(0); + List spans = SegmentHelper.getSpans(segment); + assertThat(spans.size(), is(1)); + + assertMessageSpan(spans.get(0)); + } + + private void assertMessageSpan(AbstractTracingSpan span) { + SpanAssert.assertTag(span, 0, "localhost:9092"); + SpanAssert.assertTag(span, 1, "test"); + SpanAssert.assertComponent(span, KAFKA); + SpanAssert.assertLayer(span, SpanLayer.MQ); + assertThat(span.getOperationName(), is("Kafka/test/Producer")); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java new file mode 100644 index 000000000..23f1a6ae3 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java @@ -0,0 +1,69 @@ +/* + * 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.v11; + +import java.util.ArrayList; +import java.util.List; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class ProducerConstructorInterceptorTest { + @Mock + private ProducerConfig producerConfig; + + @Mock + private ProducerConstructorInterceptor constructorInterceptor; + + private EnhancedInstance enhancedInstance = new EnhancedInstance() { + private String brokerServers; + + @Override public Object getSkyWalkingDynamicField() { + return brokerServers; + } + + @Override public void setSkyWalkingDynamicField(Object value) { + brokerServers = (String)value; + } + }; + + @Before + public void setUp() { + List mockBootstrapServers = new ArrayList(); + mockBootstrapServers.add("localhost:9092"); + mockBootstrapServers.add("localhost:19092"); + when(producerConfig.getList("bootstrap.servers")).thenReturn(mockBootstrapServers); + constructorInterceptor = new ProducerConstructorInterceptor(); + } + + @Test + public void testOnConsumer() { + constructorInterceptor.onConstruct(enhancedInstance, new Object[] {producerConfig}); + assertThat(enhancedInstance.getSkyWalkingDynamicField().toString(), is("localhost:9092;localhost:19092")); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java new file mode 100644 index 000000000..568d3644b --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java @@ -0,0 +1,59 @@ +/* + * 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.v11; + +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; + +@RunWith(MockitoJUnitRunner.class) +public class ProducerRecordConstructorInterceptorTest { + + @Mock + private ProducerRecordConstructorInterceptor constructorInterceptor; + + private EnhancedInstance enhancedInstance = new EnhancedInstance() { + private String brokerServers; + + @Override public Object getSkyWalkingDynamicField() { + return brokerServers; + } + + @Override public void setSkyWalkingDynamicField(Object value) { + brokerServers = (String)value; + } + }; + + @Before + public void setUp() { + constructorInterceptor = new ProducerRecordConstructorInterceptor(); + } + + @Test + public void testOnConsumer() { + constructorInterceptor.onConstruct(enhancedInstance, new Object[] {"test"}); + assertThat(enhancedInstance.getSkyWalkingDynamicField().toString(), is("test")); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.java new file mode 100644 index 000000000..b6d7190b7 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.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.kafka.v11; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; + +@RunWith(MockitoJUnitRunner.class) +public class SubscribeMethodInterceptorTest { + + @Mock + private SubscribeMethodInterceptor constructorInterceptor; + + private List mockTopics = new ArrayList(); + + private EnhancedInstance enhancedInstance = new EnhancedInstance() { + ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo(); + + @Override public Object getSkyWalkingDynamicField() { + return consumerEnhanceRequiredInfo; + } + + @Override public void setSkyWalkingDynamicField(Object value) { + this.consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)value; + } + }; + + @Before + public void setUp() { + mockTopics.add("test"); + mockTopics.add("test-1"); + constructorInterceptor = new SubscribeMethodInterceptor(); + } + + @Test + public void testOnConsumer() throws Throwable { + constructorInterceptor.beforeMethod(enhancedInstance, null, new Object[] {mockTopics}, new Class[] {Collection.class}, null); + ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo)enhancedInstance.getSkyWalkingDynamicField(); + assertThat(requiredInfo.getTopics(), is("test;test-1")); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java b/apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java new file mode 100644 index 000000000..afc4f8a51 --- /dev/null +++ b/apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java @@ -0,0 +1,47 @@ +/* + * 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.agent.core.context; + +import java.util.ArrayList; +import java.util.List; +import org.apache.skywalking.apm.agent.core.context.ids.DistributedTraceId; +import org.apache.skywalking.apm.agent.core.context.ids.ID; +import org.apache.skywalking.apm.agent.core.context.ids.NewDistributedTraceId; + +public enum MockContextSnapshot { + INSTANCE; + + private ContextSnapshot contextSnapshot; + + MockContextSnapshot() { + List distributedTraceIds = new ArrayList(); + distributedTraceIds.add(new NewDistributedTraceId()); + + contextSnapshot = new ContextSnapshot(new ID(1, 2, 3), 1, distributedTraceIds); + contextSnapshot.setEntryApplicationInstanceId(1); + contextSnapshot.setEntryOperationId(0); + contextSnapshot.setEntryOperationName("/for-test-entryOperationName"); + contextSnapshot.setParentOperationId(0); + contextSnapshot.setParentOperationName("/for-test-parentOperationName"); + } + + public ContextSnapshot mockContextSnapshot() { + return contextSnapshot; + } +}