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

Polish up Pulsar plugins (#375)

This commit is contained in:
pg.yang 2022-11-05 22:18:32 +08:00 committed by GitHub
parent 35a3f5a59b
commit f17a4ea54a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 223 additions and 660 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.md
View File

@ -26,6 +26,7 @@ Release Notes.
* Add RocketMQ test scenarios for version 4.3 - 4.9. No 4.0 - 4.2 release images for testing.
* Support mannual propagation of tracing context to next operators for webflux.
* Add MQ_TOPIC and MQ_BROKER tags for RocketMQ consumer's span.
* Polish up Pulsar plugins to remove unnecessary dynamic value , set peer at consumer side
#### Documentation

21
apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptor.java
View File

@ -18,32 +18,13 @@
package org.apache.skywalking.apm.plugin.pulsar;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
/**
* Interceptor of pulsar producer constructor.
* <p>
* The interceptor create {@link ProducerEnhanceRequiredInfo} which is required by instance method interceptor, So use
* it to update the skywalking dynamic field of pulsar producer enhanced instance. So that the instance methods can get
* the {@link ProducerEnhanceRequiredInfo}
*/
public class ProducerConstructorInterceptor implements InstanceConstructorInterceptor {
@Override
public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
PulsarClientImpl pulsarClient = (PulsarClientImpl) allArguments[0];
String topic = (String) allArguments[1];
ProducerEnhanceRequiredInfo producerEnhanceRequiredInfo = new ProducerEnhanceRequiredInfo();
producerEnhanceRequiredInfo.setTopic(topic);
/*
* Pulsar url can specify with specific URL or a service url provider, use pulsarClient.getLookup().getServiceUrl()
* can handle the service url provider which use a dynamic service url
*/
producerEnhanceRequiredInfo.setServiceUrl(pulsarClient.getLookup().getServiceUrl());
producerEnhanceRequiredInfo.setPropertiesInjector(new MessagePropertiesInjectorV27());
objInst.setSkyWalkingDynamicField(producerEnhanceRequiredInfo);
objInst.setSkyWalkingDynamicField(new MessagePropertiesInjectorV27());
}
}

82
apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/MockMessage.java
View File

@ -1,82 +0,0 @@
/*
* 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.pulsar;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.shade.io.netty.buffer.ByteBuf;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class MockMessage extends MessageImpl implements EnhancedInstance {
private PulsarApi.MessageMetadata.Builder msgMetadataBuilder = PulsarApi.MessageMetadata.newBuilder();
private transient Map<String, String> properties;
private Object enhancedSkyWalkingField;
public MockMessage() {
this(null, "1:1", new HashMap(), null, null);
}
public MockMessage(String topic, String msgId, Map properties, ByteBuf payload, Schema schema) {
super(topic, msgId, properties, payload, schema);
}
@Override
public PulsarApi.MessageMetadata.Builder getMessageBuilder() {
return msgMetadataBuilder;
}
public synchronized Map<String, String> getProperties() {
if (this.properties == null) {
if (this.msgMetadataBuilder.getPropertiesCount() > 0) {
Map<String, String> internalProperties = new HashMap<String, String>();
for (int i = 0; i < this.msgMetadataBuilder.getPropertiesCount(); i++) {
PulsarApi.KeyValue kv = this.msgMetadataBuilder.getProperties(i);
internalProperties.put(kv.getKey(), kv.getValue());
}
this.properties = Collections.unmodifiableMap(internalProperties);
} else {
this.properties = Collections.emptyMap();
}
}
return this.properties;
}
@Override
public String getProperty(String name) {
return this.getProperties().get(name);
}
@Override
public Object getSkyWalkingDynamicField() {
return enhancedSkyWalkingField;
}
@Override
public void setSkyWalkingDynamicField(Object value) {
this.enhancedSkyWalkingField = value;
}
}

96
apm-sniffer/apm-sdk-plugin/pulsar-2.2-2.7-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/ProducerConstructorInterceptorTest.java
View File

@ -1,96 +0,0 @@
/*
* 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.pulsar;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.Is.is;
import static org.mockito.Mockito.when;
import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.plugin.pulsar.common.MessagePropertiesInjector;
import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class ProducerConstructorInterceptorTest {
private static final String SERVICE_URL = "pulsar://localhost:6650";
private static final String TOPIC_NAME = "persistent://my-tenant/my-ns/my-topic";
private static final String HEAD_KEY = "testKey";
private static final String HEAD_VALUE = "testValue";
@Mock
private PulsarClientImpl pulsarClient;
@Mock
private LookupService lookupService;
@Mock
private CarrierItem carrierItem;
private MessageImpl<?> message;
private ProducerConstructorInterceptor constructorInterceptor;
private EnhancedInstance enhancedInstance = new EnhancedInstance() {
private ProducerEnhanceRequiredInfo requiredInfo;
@Override
public Object getSkyWalkingDynamicField() {
return requiredInfo;
}
@Override
public void setSkyWalkingDynamicField(Object value) {
this.requiredInfo = (ProducerEnhanceRequiredInfo) value;
}
};
@Before
public void setUp() {
when(lookupService.getServiceUrl()).thenReturn(SERVICE_URL);
when(pulsarClient.getLookup()).thenReturn(lookupService);
when(carrierItem.getHeadKey()).thenReturn(HEAD_KEY);
when(carrierItem.getHeadValue()).thenReturn(HEAD_VALUE);
constructorInterceptor = new ProducerConstructorInterceptor();
message = new MockMessage();
}
@Test
public void testOnConsumer() {
constructorInterceptor.onConstruct(enhancedInstance, new Object[] {
pulsarClient,
TOPIC_NAME
});
ProducerEnhanceRequiredInfo requiredInfo = (ProducerEnhanceRequiredInfo) enhancedInstance.getSkyWalkingDynamicField();
assertThat(requiredInfo.getServiceUrl(), is(SERVICE_URL));
assertThat(requiredInfo.getTopic(), is(TOPIC_NAME));
final MessagePropertiesInjector propertiesInjector = requiredInfo.getPropertiesInjector();
Assert.assertNotNull(propertiesInjector);
propertiesInjector.inject(message, carrierItem);
assertThat(message.getProperty(HEAD_KEY), is(HEAD_VALUE));
}
}

21
apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptor.java
View File

@ -18,32 +18,13 @@
package org.apache.skywalking.apm.plugin.pulsar.v28x;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
/**
* Interceptor of pulsar producer constructor.
* <p>
* The interceptor create {@link ProducerEnhanceRequiredInfo} which is required by instance method interceptor, So use
* it to update the skywalking dynamic field of pulsar producer enhanced instance. So that the instance methods can get
* the {@link ProducerEnhanceRequiredInfo}
*/
public class ProducerConstructorInterceptor implements InstanceConstructorInterceptor {
@Override
public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
PulsarClientImpl pulsarClient = (PulsarClientImpl) allArguments[0];
String topic = (String) allArguments[1];
ProducerEnhanceRequiredInfo producerEnhanceRequiredInfo = new ProducerEnhanceRequiredInfo();
producerEnhanceRequiredInfo.setTopic(topic);
/*
* Pulsar url can specify with specific URL or a service url provider, use pulsarClient.getLookup().getServiceUrl()
* can handle the service url provider which use a dynamic service url
*/
producerEnhanceRequiredInfo.setServiceUrl(pulsarClient.getLookup().getServiceUrl());
producerEnhanceRequiredInfo.setPropertiesInjector(new MessagePropertiesInjectorV28());
objInst.setSkyWalkingDynamicField(producerEnhanceRequiredInfo);
objInst.setSkyWalkingDynamicField(new MessagePropertiesInjectorV28());
}
}

98
apm-sniffer/apm-sdk-plugin/pulsar-2.8.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/pulsar/v28x/ProducerConstructorInterceptorTest.java
View File

@ -1,98 +0,0 @@
/*
* 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.pulsar.v28x;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.Is.is;
import static org.mockito.Mockito.when;
import java.nio.ByteBuffer;
import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.plugin.pulsar.common.MessagePropertiesInjector;
import org.apache.skywalking.apm.plugin.pulsar.common.ProducerEnhanceRequiredInfo;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class ProducerConstructorInterceptorTest {
private static final String SERVICE_URL = "pulsar://localhost:6650";
private static final String TOPIC_NAME = "persistent://my-tenant/my-ns/my-topic";
private static final String HEAD_KEY = "testKey";
private static final String HEAD_VALUE = "testValue";
@Mock
private PulsarClientImpl pulsarClient;
@Mock
private LookupService lookupService;
@Mock
private CarrierItem carrierItem;
private MessageImpl<?> message;
private ProducerConstructorInterceptor constructorInterceptor;
private EnhancedInstance enhancedInstance = new EnhancedInstance() {
private ProducerEnhanceRequiredInfo requiredInfo;
@Override
public Object getSkyWalkingDynamicField() {
return requiredInfo;
}
@Override
public void setSkyWalkingDynamicField(Object value) {
this.requiredInfo = (ProducerEnhanceRequiredInfo) value;
}
};
@Before
public void setUp() {
when(lookupService.getServiceUrl()).thenReturn(SERVICE_URL);
when(pulsarClient.getLookup()).thenReturn(lookupService);
when(carrierItem.getHeadKey()).thenReturn(HEAD_KEY);
when(carrierItem.getHeadValue()).thenReturn(HEAD_VALUE);
constructorInterceptor = new ProducerConstructorInterceptor();
message = MessageImpl.create(new MessageMetadata(), ByteBuffer.allocate(1), null);
}
@Test
public void testOnConsumer() {
constructorInterceptor.onConstruct(enhancedInstance, new Object[] {
pulsarClient,
TOPIC_NAME
});
ProducerEnhanceRequiredInfo requiredInfo = (ProducerEnhanceRequiredInfo) enhancedInstance.getSkyWalkingDynamicField();
assertThat(requiredInfo.getServiceUrl(), is(SERVICE_URL));
assertThat(requiredInfo.getTopic(), is(TOPIC_NAME));
final MessagePropertiesInjector propertiesInjector = requiredInfo.getPropertiesInjector();
Assert.assertNotNull(propertiesInjector);
propertiesInjector.inject(message, carrierItem);
assertThat(message.getProperty(HEAD_KEY), is(HEAD_VALUE));
}
}

20
apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptor.java
View File

@ -18,33 +18,15 @@
package org.apache.skywalking.apm.plugin.pulsar.common;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
/**
* Interceptor of pulsar consumer constructor.
* <p>
* The interceptor create {@link ConsumerEnhanceRequiredInfo} which is required by instance method interceptor, So use
* it to update the skywalking dynamic field of pulsar consumer enhanced instance. So that the instance methods can get
* the {@link ConsumerEnhanceRequiredInfo}
*/
public class ConsumerConstructorInterceptor implements InstanceConstructorInterceptor {
@Override
public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
PulsarClientImpl pulsarClient = (PulsarClientImpl) allArguments[0];
String topic = (String) allArguments[1];
ConsumerConfigurationData consumerConfigurationData = (ConsumerConfigurationData) allArguments[2];
ConsumerEnhanceRequiredInfo requireInfo = new ConsumerEnhanceRequiredInfo();
/*
* Pulsar url can specify with specific URL or a service url provider, use pulsarClient.getLookup().getServiceUrl()
* can handle the service url provider which use a dynamic service url
*/
requireInfo.setServiceUrl(pulsarClient.getLookup().getServiceUrl());
requireInfo.setTopic(topic);
requireInfo.setSubscriptionName(consumerConfigurationData.getSubscriptionName());
objInst.setSkyWalkingDynamicField(requireInfo);
objInst.setSkyWalkingDynamicField(consumerConfigurationData.getSubscriptionName());
}
}

65
apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerEnhanceRequiredInfo.java
View File

@ -1,65 +0,0 @@
/*
* 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.pulsar.common;
/**
* Pulsar consumer enhance required info is required by consumer enhanced object method interceptor
*/
public class ConsumerEnhanceRequiredInfo {
/**
* service url of the consumer
*/
private String serviceUrl;
/**
* topic name of the consumer
*/
private String topic;
/**
* subscription name of the consumer
*/
private String subscriptionName;
public String getServiceUrl() {
return serviceUrl;
}
public void setServiceUrl(String serviceUrl) {
this.serviceUrl = serviceUrl;
}
public String getTopic() {
return topic;
}
public void setTopic(String topic) {
this.topic = topic;
}
public String getSubscriptionName() {
return subscriptionName;
}
public void setSubscriptionName(String subscriptionName) {
this.subscriptionName = subscriptionName;
}
}

65
apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/ProducerEnhanceRequiredInfo.java
View File

@ -1,65 +0,0 @@
/*
* 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.pulsar.common;
/**
* Pulsar producer enhance required info is required by producer enhanced object method interceptor
*/
public class ProducerEnhanceRequiredInfo {
/**
* service url of the pulsar producer
*/
private String serviceUrl;
/**
* topic name of the pulsar producer
*/
private String topic;
/**
* Message properties injector
*/
private MessagePropertiesInjector propertiesInjector;
public MessagePropertiesInjector getPropertiesInjector() {
return propertiesInjector;
}
public void setPropertiesInjector(
MessagePropertiesInjector propertiesInjector) {
this.propertiesInjector = propertiesInjector;
}
public String getTopic() {
return topic;
}
public void setTopic(String topic) {
this.topic = topic;
}
public String getServiceUrl() {
return serviceUrl;
}
public void setServiceUrl(String serviceUrl) {
this.serviceUrl = serviceUrl;
}
}

34
apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptor.java
View File

@ -18,7 +18,9 @@
package org.apache.skywalking.apm.plugin.pulsar.common;
import java.lang.reflect.Method;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
@ -30,15 +32,13 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceM
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import java.lang.reflect.Method;
/**
* Interceptor for pulsar consumer enhanced instance
* <p>
* Here is the intercept process steps:
*
* <pre>
* 1. Get the @{@link ConsumerEnhanceRequiredInfo} and record the service url, topic name and subscription name
* 1. Record the service url, topic name and subscription name through this(ConsumerImpl)
* 2. Create the entry span when call <code>messageProcessed</code> method
* 3. Extract all the <code>Trace Context</code> when call <code>messageProcessed</code> method
* 4. Capture trace context and set into SkyWalkingDynamic field if consumer has a message listener when <code>messageProcessed</code> method finished
@ -52,9 +52,11 @@ public class PulsarConsumerInterceptor implements InstanceMethodsAroundIntercept
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
MethodInterceptResult result) throws Throwable {
if (allArguments[0] != null) {
ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField();
ConsumerImpl consumer = (ConsumerImpl) objInst;
final String serviceUrl = consumer.getClient().getLookup().getServiceUrl();
Message msg = (Message) allArguments[0];
ContextCarrier carrier = new ContextCarrier();
CarrierItem next = carrier.items();
@ -62,29 +64,31 @@ public class PulsarConsumerInterceptor implements InstanceMethodsAroundIntercept
next = next.next();
next.setHeadValue(msg.getProperty(next.getHeadKey()));
}
AbstractSpan activeSpan = ContextManager.createEntrySpan(OPERATE_NAME_PREFIX + requiredInfo.getTopic() + CONSUMER_OPERATE_NAME + requiredInfo
.getSubscriptionName(), carrier);
AbstractSpan activeSpan = ContextManager.createEntrySpan(
OPERATE_NAME_PREFIX + consumer.getTopic() + CONSUMER_OPERATE_NAME + objInst.getSkyWalkingDynamicField(),
carrier
);
activeSpan.setComponent(ComponentsDefine.PULSAR_CONSUMER);
SpanLayer.asMQ(activeSpan);
Tags.MQ_BROKER.set(activeSpan, requiredInfo.getServiceUrl());
Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopic());
Tags.MQ_BROKER.set(activeSpan, serviceUrl);
Tags.MQ_TOPIC.set(activeSpan, consumer.getTopic());
activeSpan.setPeer(serviceUrl);
}
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
Object ret) throws Throwable {
if (allArguments[0] != null) {
final ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst
.getSkyWalkingDynamicField();
ConsumerImpl consumer = (ConsumerImpl) objInst;
EnhancedInstance msg = (EnhancedInstance) allArguments[0];
MessageEnhanceRequiredInfo messageEnhanceRequiredInfo = (MessageEnhanceRequiredInfo) msg
.getSkyWalkingDynamicField();
.getSkyWalkingDynamicField();
if (messageEnhanceRequiredInfo == null) {
messageEnhanceRequiredInfo = new MessageEnhanceRequiredInfo();
msg.setSkyWalkingDynamicField(messageEnhanceRequiredInfo);
}
messageEnhanceRequiredInfo.setTopic(requiredInfo.getTopic());
messageEnhanceRequiredInfo.setTopic(consumer.getTopic());
messageEnhanceRequiredInfo.setContextSnapshot(ContextManager.capture());
ContextManager.stopSpan();
}
@ -93,7 +97,7 @@ public class PulsarConsumerInterceptor implements InstanceMethodsAroundIntercept
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
Class<?>[] argumentsTypes, Throwable t) {
if (allArguments[0] != null) {
ContextManager.activeSpan().log(t);
}

18
apm-sniffer/apm-sdk-plugin/pulsar-common/src/main/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptor.java
View File

@ -20,6 +20,7 @@ package org.apache.skywalking.apm.plugin.pulsar.common;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
@ -40,7 +41,7 @@ import java.lang.reflect.Method;
* Here is the intercept process steps:
*
* <pre>
* 1. Get the {@link ProducerEnhanceRequiredInfo} and record the service url, topic name
* 1. Record the service url, topic name through this(ProducerImpl)
* 2. Create the exit span when the producer invoke <code>sendAsync</code> method
* 3. Inject the context to {@link Message#getProperties()}
* 4. Create {@link SendCallbackEnhanceRequiredInfo} with <code>ContextManager.capture()</code> and set the
@ -57,19 +58,18 @@ public class PulsarProducerInterceptor implements InstanceMethodsAroundIntercept
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
if (allArguments[0] != null) {
ProducerEnhanceRequiredInfo requiredInfo = (ProducerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField();
ContextCarrier contextCarrier = new ContextCarrier();
String topicName = requiredInfo.getTopic();
AbstractSpan activeSpan = ContextManager.createExitSpan(OPERATE_NAME_PREFIX + topicName + PRODUCER_OPERATE_NAME_SUFFIX, contextCarrier, requiredInfo
.getServiceUrl());
Tags.MQ_BROKER.set(activeSpan, requiredInfo.getServiceUrl());
Tags.MQ_TOPIC.set(activeSpan, topicName);
ProducerImpl producer = (ProducerImpl) objInst;
final String serviceUrl = producer.getClient().getLookup().getServiceUrl();
AbstractSpan activeSpan = ContextManager.createExitSpan(OPERATE_NAME_PREFIX + producer.getTopic() + PRODUCER_OPERATE_NAME_SUFFIX, contextCarrier, serviceUrl);
Tags.MQ_BROKER.set(activeSpan, serviceUrl);
Tags.MQ_TOPIC.set(activeSpan, producer.getTopic());
contextCarrier.extensionInjector().injectSendingTimestamp();
SpanLayer.asMQ(activeSpan);
activeSpan.setComponent(ComponentsDefine.PULSAR_PRODUCER);
CarrierItem next = contextCarrier.items();
MessageImpl msg = (MessageImpl) allArguments[0];
final MessagePropertiesInjector propertiesInjector = requiredInfo.getPropertiesInjector();
MessagePropertiesInjector propertiesInjector = (MessagePropertiesInjector) objInst.getSkyWalkingDynamicField();
if (propertiesInjector != null) {
while (next.hasNext()) {
next = next.next();
@ -83,7 +83,7 @@ public class PulsarProducerInterceptor implements InstanceMethodsAroundIntercept
ContextSnapshot snapshot = ContextManager.capture();
if (null != snapshot) {
SendCallbackEnhanceRequiredInfo callbackRequiredInfo = new SendCallbackEnhanceRequiredInfo();
callbackRequiredInfo.setTopic(topicName);
callbackRequiredInfo.setTopic(producer.getTopic());
callbackRequiredInfo.setContextSnapshot(snapshot);
callbackInstance.setSkyWalkingDynamicField(callbackRequiredInfo);
}

88
apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/ConsumerConstructorInterceptorTest.java
View File

@ -1,88 +0,0 @@
/*
* 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.pulsar.common;
import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
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 {
private static final String SERVICE_URL = "pulsar://localhost:6650";
private static final String TOPIC_NAME = "persistent://my-tenant/my-ns/my-topic";
private static final String SUBSCRIPTION_NAME = "my-sub";
@Mock
private PulsarClientImpl pulsarClient;
@Mock
private LookupService lookupService;
@Mock
private ConsumerConfigurationData consumerConfigurationData;
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() {
when(lookupService.getServiceUrl()).thenReturn(SERVICE_URL);
when(pulsarClient.getLookup()).thenReturn(lookupService);
when(consumerConfigurationData.getSubscriptionName()).thenReturn(SUBSCRIPTION_NAME);
constructorInterceptor = new ConsumerConstructorInterceptor();
}
@Test
public void testOnConsumer() {
constructorInterceptor.onConstruct(enhancedInstance, new Object[] {
pulsarClient,
TOPIC_NAME,
consumerConfigurationData
});
ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) enhancedInstance.getSkyWalkingDynamicField();
assertThat(requiredInfo.getServiceUrl(), is(SERVICE_URL));
assertThat(requiredInfo.getTopic(), is(TOPIC_NAME));
assertThat(requiredInfo.getSubscriptionName(), is(SUBSCRIPTION_NAME));
}
}

57
apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockConsumerImpl.java Normal file
View File

@ -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.pulsar.common;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.pulsar.client.impl.ConsumerInterceptors;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
public class MockConsumerImpl extends ConsumerImpl implements EnhancedInstance {
protected MockConsumerImpl(final PulsarClientImpl client,
final String topic,
final ConsumerConfigurationData conf,
final ExecutorService listenerExecutor,
final int partitionIndex,
final CompletableFuture subscribeFuture,
final MessageId startMessageId,
final Schema schema,
final ConsumerInterceptors interceptors,
final long backoffIntervalNanos,
final long maxBackoffIntervalNanos) {
super(client, topic, conf, listenerExecutor, partitionIndex, subscribeFuture, null, startMessageId,
schema, interceptors, backoffIntervalNanos, maxBackoffIntervalNanos
);
}
@Override
public Object getSkyWalkingDynamicField() {
return null;
}
@Override
public void setSkyWalkingDynamicField(final Object value) {
}
}

48
apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/MockProducerImpl.java Normal file
View File

@ -0,0 +1,48 @@
/*
* 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.pulsar.common;
import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.pulsar.client.impl.ProducerInterceptors;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
public class MockProducerImpl extends ProducerImpl implements EnhancedInstance {
public MockProducerImpl(final PulsarClientImpl client,
final String topic,
final ProducerConfigurationData conf,
final CompletableFuture producerCreatedFuture,
final int partitionIndex,
final Schema schema,
final ProducerInterceptors interceptors) {
super(client, topic, conf, producerCreatedFuture, partitionIndex, schema, interceptors);
}
@Override
public Object getSkyWalkingDynamicField() {
return null;
}
@Override
public void setSkyWalkingDynamicField(final Object value) {
}
}

51
apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerInterceptorTest.java
View File

@ -18,6 +18,9 @@
package org.apache.skywalking.apm.plugin.pulsar.common;
import java.util.List;
import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.skywalking.apm.agent.core.context.SW8CarrierItem;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
@ -37,11 +40,10 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.modules.junit4.PowerMockRunnerDelegate;
import java.util.List;
import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.PULSAR_CONSUMER;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertNotNull;
@ -57,45 +59,32 @@ public class PulsarConsumerInterceptorTest {
@Rule
public AgentServiceRule serviceRule = new AgentServiceRule();
private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo;
private MessageEnhanceRequiredInfo messageEnhanceRequiredInfo;
private PulsarConsumerInterceptor consumerInterceptor;
private MockMessage msg;
private EnhancedInstance consumerInstance = new EnhancedInstance() {
@Override
public Object getSkyWalkingDynamicField() {
return consumerEnhanceRequiredInfo;
return "my-sub";
}
@Override
public void setSkyWalkingDynamicField(Object value) {
consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo) value;
}
};
@Before
public void setUp() {
consumerInterceptor = new PulsarConsumerInterceptor();
consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo();
consumerEnhanceRequiredInfo.setTopic("persistent://my-tenant/my-ns/my-topic");
consumerEnhanceRequiredInfo.setServiceUrl("pulsar://localhost:6650");
consumerEnhanceRequiredInfo.setSubscriptionName("my-sub");
msg = new MockMessage();
msg.getMessageBuilder()
.addProperties(PulsarApi.KeyValue.newBuilder()
.setKey(SW8CarrierItem.HEADER_NAME)
.setValue("1-My40LjU=-MS4yLjM=-3-c2VydmljZQ==-aW5zdGFuY2U=-L2FwcA==-MTI3LjAuMC4xOjgwODA="));
messageEnhanceRequiredInfo = new MessageEnhanceRequiredInfo();
msg.setSkyWalkingDynamicField(messageEnhanceRequiredInfo);
}
@Test
public void testConsumerWithNullMessage() throws Throwable {
PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
consumerInterceptor.beforeMethod(consumerInstance, null, new Object[] {null}, new Class[0], null);
consumerInterceptor.afterMethod(consumerInstance, null, new Object[] {null}, new Class[0], null);
@ -105,8 +94,10 @@ public class PulsarConsumerInterceptorTest {
@Test
public void testConsumerWithMessage() throws Throwable {
consumerInterceptor.beforeMethod(consumerInstance, null, new Object[] {msg}, new Class[0], null);
consumerInterceptor.afterMethod(consumerInstance, null, new Object[] {msg}, new Class[0], null);
EnhancedInstance enhancedInstance = mockConsumer();
PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
consumerInterceptor.beforeMethod(enhancedInstance, null, new Object[] {msg}, new Class[0], null);
consumerInterceptor.afterMethod(enhancedInstance, null, new Object[] {msg}, new Class[0], null);
List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
assertThat(traceSegments.size(), is(1));
@ -122,8 +113,10 @@ public class PulsarConsumerInterceptorTest {
@Test
public void testConsumerWithMessageListener() throws Throwable {
consumerInterceptor.beforeMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
consumerInterceptor.afterMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
EnhancedInstance enhancedInstance = mockConsumer();
PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
consumerInterceptor.beforeMethod(enhancedInstance, null, new Object[]{msg}, new Class[0], null);
consumerInterceptor.afterMethod(enhancedInstance, null, new Object[]{msg}, new Class[0], null);
List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
assertThat(traceSegments.size(), is(1));
@ -137,8 +130,6 @@ public class PulsarConsumerInterceptorTest {
assertConsumerSpan(spans.get(0));
final MessageEnhanceRequiredInfo requiredInfo = (MessageEnhanceRequiredInfo) msg.getSkyWalkingDynamicField();
assertThat(requiredInfo.getTopic(), is(
((ConsumerEnhanceRequiredInfo) consumerInstance.getSkyWalkingDynamicField()).getTopic()));
assertNotNull(requiredInfo.getContextSnapshot());
}
@ -155,4 +146,16 @@ public class PulsarConsumerInterceptorTest {
MatcherAssert.assertThat(SegmentRefHelper.getSpanId(ref), is(3));
MatcherAssert.assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("3.4.5"));
}
private EnhancedInstance mockConsumer() throws Throwable {
EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockConsumerImpl.class);
final LookupService lookup = PowerMockito.mock(LookupService.class);
final PulsarClientImpl client = PowerMockito.mock(PulsarClientImpl.class);
PowerMockito.when(lookup, "getServiceUrl").thenReturn("pulsar://localhost:6650");
PowerMockito.when(client, "getLookup").thenReturn(lookup);
PowerMockito.when(pulsarProducerInstance, "getClient").thenReturn(client);
PowerMockito.when(pulsarProducerInstance, "getTopic").thenReturn("persistent://my-tenant/my-ns/my-topic");
PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn("my-sub");
return pulsarProducerInstance;
}
}

52
apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarConsumerListenerInterceptorTest.java
View File

@ -18,7 +18,11 @@
package org.apache.skywalking.apm.plugin.pulsar.common;
import java.util.List;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.skywalking.apm.agent.core.context.SW8CarrierItem;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
@ -39,11 +43,10 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.modules.junit4.PowerMockRunnerDelegate;
import java.util.List;
import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.PULSAR_CONSUMER;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertNotNull;
@ -53,7 +56,6 @@ import static org.junit.Assert.assertThat;
@RunWith(PowerMockRunner.class)
@PowerMockRunnerDelegate(TracingSegmentRunner.class)
public class PulsarConsumerListenerInterceptorTest {
@Rule
public AgentServiceRule serviceRule = new AgentServiceRule();
@SegmentStoragePoint
@ -70,32 +72,13 @@ public class PulsarConsumerListenerInterceptorTest {
};
private PulsarConsumerListenerInterceptor consumerListenerInterceptor;
private MockMessage msg;
private PulsarConsumerInterceptor consumerInterceptor;
private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo;
private final EnhancedInstance consumerInstance = new EnhancedInstance() {
@Override
public Object getSkyWalkingDynamicField() {
return consumerEnhanceRequiredInfo;
}
@Override
public void setSkyWalkingDynamicField(Object value) {
consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo) value;
}
};
private MockConsumer consumer;
private MessageListener messageListener;
@Before
public void setUp() {
consumerInterceptor = new PulsarConsumerInterceptor();
consumerListenerInterceptor = new PulsarConsumerListenerInterceptor();
messageListener = (consumer, message) -> message.getTopicName();
consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo();
consumerEnhanceRequiredInfo.setTopic("persistent://my-tenant/my-ns/my-topic");
consumerEnhanceRequiredInfo.setServiceUrl("pulsar://localhost:6650");
consumerEnhanceRequiredInfo.setSubscriptionName("my-sub");
msg = new MockMessage();
msg.getMessageBuilder()
.addProperties(PulsarApi.KeyValue.newBuilder()
@ -134,16 +117,17 @@ public class PulsarConsumerListenerInterceptorTest {
@Test
public void testWithMessageListenerHasRequiredInfo() throws Throwable {
consumerInterceptor.beforeMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
consumerInterceptor.afterMethod(consumerInstance, null, new Object[]{msg}, new Class[0], null);
PulsarConsumerInterceptor consumerInterceptor = new PulsarConsumerInterceptor();
EnhancedInstance consumer = mockConsumer();
consumerInterceptor.beforeMethod(consumer, null, new Object[]{msg}, new Class[0], null);
consumerInterceptor.afterMethod(consumer, null, new Object[]{msg}, new Class[0], null);
consumerListenerInterceptor
.beforeMethod(consumerConfigurationDataInstance, null, new Object[0], new Class[0], null);
final MessageListener enhancedMessageListener = (MessageListener) consumerListenerInterceptor
.afterMethod(consumerConfigurationDataInstance, null, new Object[0], new Class[0],
this.messageListener);
assertNotNull(enhancedMessageListener);
enhancedMessageListener.received(consumer, msg);
enhancedMessageListener.received((Consumer) consumer, msg);
List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
assertThat(traceSegments.size(), is(2));
@ -156,6 +140,22 @@ public class PulsarConsumerListenerInterceptorTest {
assertConsumerSpan(spans.get(0));
}
private EnhancedInstance mockConsumer() {
EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockConsumerImpl.class);
final LookupService lookup = PowerMockito.mock(LookupService.class);
final PulsarClientImpl client = PowerMockito.mock(PulsarClientImpl.class);
try {
PowerMockito.when(lookup, "getServiceUrl").thenReturn("pulsar://localhost:6650");
PowerMockito.when(client, "getLookup").thenReturn(lookup);
PowerMockito.when(pulsarProducerInstance, "getClient").thenReturn(client);
PowerMockito.when(pulsarProducerInstance, "getTopic").thenReturn("persistent://my-tenant/my-ns/my-topic");
PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn("my-sub");
} catch (Exception e) {
throw new RuntimeException(e);
}
return pulsarProducerInstance;
}
private void assertConsumerSpan(AbstractTracingSpan span) {
SpanAssert.assertLayer(span, SpanLayer.MQ);
SpanAssert.assertComponent(span, PULSAR_CONSUMER);

34
apm-sniffer/apm-sdk-plugin/pulsar-common/src/test/java/org/apache/skywalking/apm/plugin/pulsar/common/PulsarProducerInterceptorTest.java
View File

@ -18,7 +18,10 @@
package org.apache.skywalking.apm.plugin.pulsar.common;
import java.util.List;
import org.apache.pulsar.client.impl.LookupService;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
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;
@ -33,11 +36,10 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.modules.junit4.PowerMockRunnerDelegate;
import java.util.List;
import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.PULSAR_PRODUCER;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
@ -57,21 +59,6 @@ public class PulsarProducerInterceptorTest {
private Object[] arguments;
private Class[] argumentType;
private EnhancedInstance pulsarProducerInstance = new EnhancedInstance() {
@Override
public Object getSkyWalkingDynamicField() {
ProducerEnhanceRequiredInfo requiredInfo = new ProducerEnhanceRequiredInfo();
requiredInfo.setTopic("persistent://my-tenant/my-ns/my-topic");
requiredInfo.setServiceUrl("pulsar://localhost:6650");
return requiredInfo;
}
@Override
public void setSkyWalkingDynamicField(Object value) {
}
};
private MessageImpl msg = new MockMessage();
@Before
@ -86,6 +73,16 @@ public class PulsarProducerInterceptorTest {
@Test
public void testSendMessage() throws Throwable {
EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockProducerImpl.class);
final LookupService lookup = PowerMockito.mock(LookupService.class);
final PulsarClientImpl client = PowerMockito.mock(PulsarClientImpl.class);
PowerMockito.when(lookup, "getServiceUrl").thenReturn("pulsar://localhost:6650");
PowerMockito.when(client, "getLookup").thenReturn(lookup);
PowerMockito.when(pulsarProducerInstance, "getClient").thenReturn(client);
PowerMockito.when(pulsarProducerInstance, "getTopic").thenReturn("persistent://my-tenant/my-ns/my-topic");
PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn((MessagePropertiesInjector) (message, carrierItem) -> {
});
producerInterceptor.beforeMethod(pulsarProducerInstance, null, arguments, argumentType, null);
producerInterceptor.afterMethod(pulsarProducerInstance, null, arguments, argumentType, null);
@ -101,6 +98,9 @@ public class PulsarProducerInterceptorTest {
@Test
public void testSendWithNullMessage() throws Throwable {
EnhancedInstance pulsarProducerInstance = PowerMockito.mock(MockProducerImpl.class);
PowerMockito.when(pulsarProducerInstance, "getSkyWalkingDynamicField").thenReturn((MessagePropertiesInjector) (message, carrierItem) -> {
});
producerInterceptor.beforeMethod(pulsarProducerInstance, null, new Object[] {null}, argumentType, null);
producerInterceptor.afterMethod(pulsarProducerInstance, null, new Object[] {null}, argumentType, null);
List<TraceSegment> traceSegmentList = segmentStorage.getTraceSegments();

32
test/plugin/scenarios/pulsar-scenario/config/expectedData.yaml
View File

@ -28,9 +28,9 @@ segmentItems:
componentId: 73
isError: false
spanType: Exit
peer: not null
peer: not blank
tags:
- { key: mq.broker, value: not null }
- { key: mq.broker, value: not blank }
- { key: mq.topic, value: test }
skipAnalysis: 'false'
- operationName: Pulsar/testMultiPartition/Producer
@ -42,10 +42,10 @@ segmentItems:
componentId: 73
isError: false
spanType: Exit
peer: not null
peer: not blank
skipAnalysis: false
tags:
- { key: mq.broker, value: not null }
- { key: mq.broker, value: not blank }
- { key: mq.topic, value: testMultiPartition }
- operationName: GET:/case/pulsar-case
parentSpanId: -1
@ -92,10 +92,10 @@ segmentItems:
componentId: 74
isError: false
spanType: Entry
peer: ''
peer: not blank
tags:
- { key: transmission.latency, value: not null }
- { key: mq.broker, value: not null }
- { key: transmission.latency, value: not blank }
- { key: mq.broker, value: not blank }
- { key: mq.topic, value: test }
refs:
- { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null, refType: CrossProcess,
@ -113,11 +113,11 @@ segmentItems:
componentId: 74
isError: false
spanType: Entry
peer: ''
peer: not blank
skipAnalysis: 'false'
tags:
- { key: transmission.latency, value: not null }
- { key: mq.broker, value: not null }
- { key: transmission.latency, value: not blank }
- { key: mq.broker, value: not blank }
- { key: mq.topic, value: test }
refs:
- { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null,
@ -173,10 +173,10 @@ segmentItems:
componentId: 74
isError: false
spanType: Entry
peer: ''
peer: not blank
tags:
- { key: transmission.latency, value: not null }
- { key: mq.broker, value: not null }
- { key: transmission.latency, value: not blank }
- { key: mq.broker, value: not blank }
- { key: mq.topic, value: testMultiPartition }
refs:
- { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null, refType: CrossProcess,
@ -194,11 +194,11 @@ segmentItems:
componentId: 74
isError: false
spanType: Entry
peer: ''
peer: not blank
skipAnalysis: 'false'
tags:
- { key: transmission.latency, value: not null }
- { key: mq.broker, value: not null }
- { key: transmission.latency, value: not blank }
- { key: mq.broker, value: not blank }
- { key: mq.topic, value: testMultiPartition }
refs:
- { parentEndpoint: GET:/case/pulsar-case, networkAddress: not null,