kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add testcase

This commit is contained in:
ascrutae 2018-01-23 20:03:50 +08:00
parent 5fab1fada0
commit 906a2c4d05
16 changed files with 785 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractSpan.java
View File

@ -116,4 +116,6 @@ public interface AbstractSpan {
* @param ref segment ref
*/
void ref(TraceSegmentRef ref);
AbstractSpan start(long starttime);
}

6
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java
View File

@ -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();

4
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/NoopSpan.java
View File

@ -102,4 +102,8 @@ public class NoopSpan implements AbstractSpan {
@Override public void ref(TraceSegmentRef ref) {
}
@Override public AbstractSpan start(long startTime) {
return this;
}
}

5
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md Normal file
View File

@ -0,0 +1,5 @@
# Kafka Plugin
##
## Buired Point

13
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java
View File

@ -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<String> brokerServers) {
this.brokerServers = StringUtil.join(',', brokerServers.toArray(new String[0]));
this.brokerServers = StringUtil.join(';', brokerServers.toArray(new String[0]));
}
public void setTopics(Collection<String> 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;
}
}

47
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java
View File

@ -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<TopicPartition, List<ConsumerRecord<?, ?>>> records = (Map<TopicPartition, List<ConsumerRecord<?, ?>>>)ret;
for (List<ConsumerRecord<?, ?>> 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<Header> 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<ConsumerRecord<?, ?>> consumerRecords : records.values()) {
for (ConsumerRecord<?, ?> record : consumerRecords) {
ContextCarrier contextCarrier = new ContextCarrier();
CarrierItem next = contextCarrier.items();
while (next.hasNext()) {
next = next.next();
Iterator<Header> 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;
}

13
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java
View File

@ -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);

2
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java
View File

@ -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])));
}
}

147
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java Normal file
View File

@ -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<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
assertThat(traceSegments.size(), is(1));
TraceSegment traceSegment = traceSegments.get(0);
List<AbstractTracingSpan> 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<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
assertThat(traceSegments.size(), is(1));
TraceSegment traceSegment = traceSegments.get(0);
List<AbstractTracingSpan> 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<TraceSegmentRef> 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"));
}
}

73
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java Normal file
View File

@ -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<String> mockBootstrapServers = new ArrayList<String>();
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"));
}
}

140
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java Normal file
View File

@ -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<TopicPartition, List<ConsumerRecord>> messages;
@Before
public void setUp() {
consumerInterceptor = new KafkaConsumerInterceptor();
consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo();
List<String> topics = new ArrayList<String>();
topics.add("test");
topics.add("test-1");
consumerEnhanceRequiredInfo.setTopics(topics);
List<String> brokers = new ArrayList<String>();
brokers.add("localhost:9092");
brokers.add("localhost:19092");
consumerEnhanceRequiredInfo.setBrokerServers(brokers);
messages = new HashMap<TopicPartition, List<ConsumerRecord>>();
TopicPartition topicPartition = new TopicPartition("test", 1);
List<ConsumerRecord> records = new ArrayList<ConsumerRecord>();
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<TopicPartition, List<ConsumerRecord>>());
List<TraceSegment> 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<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
assertThat(traceSegments.size(), is(1));
TraceSegment traceSegment = traceSegments.get(0);
List<TraceSegmentRef> refs = traceSegment.getRefs();
assertThat(refs.size(), is(1));
assertTraceSegmentRef(refs.get(0));
List<AbstractTracingSpan> 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"));
}
}

115
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java Normal file
View File

@ -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<TraceSegment> traceSegmentList = segmentStorage.getTraceSegments();
assertThat(traceSegmentList.size(), is(1));
TraceSegment segment = traceSegmentList.get(0);
List<AbstractTracingSpan> 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"));
}
}

69
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java Normal file
View File

@ -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<String> mockBootstrapServers = new ArrayList<String>();
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"));
}
}

59
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java Normal file
View File

@ -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"));
}
}

67
apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.java Normal file
View File

@ -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<String> mockTopics = new ArrayList<String>();
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"));
}
}

47
apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java Normal file
View File

@ -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<DistributedTraceId> distributedTraceIds = new ArrayList<DistributedTraceId>();
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;
}
}