Polish up kafka,nats,activemq,rabbitmq plugins to set peer at comsumer side (#377)

Author: pg-yang

CHANGES.md

@@ -27,6 +27,10 @@ Release Notes.
 * 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 
+* Polish Kafka plugin to set peer at the consumer side.
+* Polish NATS plugin to set peer at the consumer side.
+* Polish ActiveMQ plugin to set peer at the consumer side.
+* Polish RabbitMQ plugin to set peer at the consumer side.
 
 #### Documentation
 

apm-sniffer/apm-sdk-plugin/activemq-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/activemq/MessageConsumerDequeueInterceptor.java

@@ -83,6 +83,7 @@ public Object afterMethod(final EnhancedInstance objInst,
         if (activeSpan == null) {
             return ret;
         }
+        activeSpan.setPeer(url);
         activeSpan.setComponent(ComponentsDefine.ACTIVEMQ_CONSUMER);
         SpanLayer.asMQ(activeSpan);
         CarrierItem next = contextCarrier.items();

apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaConsumerInterceptor.java

@@ -79,7 +79,7 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA
             SpanLayer.asMQ(activeSpan);
             Tags.MQ_BROKER.set(activeSpan, requiredInfo.getBrokerServers());
             Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopics());
-
+            activeSpan.setPeer(requiredInfo.getBrokerServers());
             for (List<ConsumerRecord<?, ?>> consumerRecords : records.values()) {
                 for (ConsumerRecord<?, ?> record : consumerRecords) {
                     ContextCarrier contextCarrier = new ContextCarrier();

apm-sniffer/apm-sdk-plugin/nats-2.14.x-2.15.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/DeliverReplyInterceptor.java

@@ -36,7 +36,9 @@ public class DeliverReplyInterceptor implements InstanceMethodsAroundInterceptor
     @Override
     public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
         final AbstractSpan entrySpan = createEntrySpan((Message) allArguments[0]);
-        Tags.MQ_BROKER.set(entrySpan, buildServers((Connection) objInst));
+        final String servers = buildServers((Connection) objInst);
+        Tags.MQ_BROKER.set(entrySpan, servers);
+        entrySpan.setPeer(servers);
     }
 
     @Override

apm-sniffer/apm-sdk-plugin/nats-2.14.x-2.15.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/NatsCommons.java

@@ -108,6 +108,7 @@ static MessageHandler buildTraceMsgHandler(String servers, MessageHandler msgHan
             }
             AbstractSpan span = NatsCommons.createEntrySpan(msg);
             Tags.MQ_BROKER.set(span, servers);
+            span.setPeer(servers);
             try {
                 msgHandler.onMessage(msg);
             } catch (Exception e) {

apm-sniffer/apm-sdk-plugin/nats-2.14.x-2.15.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/SubscriptionNextMsgInterceptor.java

@@ -45,6 +45,7 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA
         Message msg = (Message) ret;
         AbstractSpan span = createEntrySpan(msg);
         Tags.MQ_BROKER.set(span , servers);
+        span.setPeer(servers);
         // Close the span immediately , as no chance to trace what user want to do
         ContextManager.stopSpan(span);
         return ret;

apm-sniffer/apm-sdk-plugin/rabbitmq-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rabbitmq/TracerConsumer.java

@@ -82,6 +82,7 @@ public void handleDelivery(final String consumerTag,
         Tags.MQ_TOPIC.set(activeSpan, envelope.getExchange());
         Tags.MQ_QUEUE.set(activeSpan, envelope.getRoutingKey());
         activeSpan.setComponent(ComponentsDefine.RABBITMQ_CONSUMER);
+        activeSpan.setPeer(serverUrl);
         SpanLayer.asMQ(activeSpan);
         CarrierItem next = contextCarrier.items();
         while (next.hasNext()) {

docs/en/setup/service-agent/java-agent/Java-Plugin-Development-Guide.md

@@ -230,6 +230,8 @@ SkyWalking analysis MQ performance related metrics through the following tags.
 * `mq.topic`   indicates MQ topic name , It's optional as some MQ don't hava concept of `topic`
 * `transmission.latency` The transmission latency from consumer to producer. Usually you needn't to record this tag manually, instead to call  `contextCarrier.extensionInjector().injectSendingTimestamp();` to record tag `sendingTimestamp` on producer side , and SkyWalking would record this tag on consumer side if `sw8-x` context carrier(from producer side) contains `sendingTimestamp`
 
+Notice , you should set `peer` at both sides(producer and consumer). And the value of peer should represent the MQ server cluster.
+
 
 ### Advanced APIs
 

test/plugin/scenarios/activemq-scenario/config/expectedData.yaml

@@ -28,7 +28,7 @@ segmentItems:
       componentId: 45
       isError: false
       spanType: Exit
-      peer: not null
+      peer: not blank
       tags:
       - {key: mq.broker, value: not blank}
       - {key: mq.queue, value: test}
@@ -59,7 +59,7 @@ segmentItems:
       componentId: 46
       isError: false
       spanType: Entry
-      peer: ''
+      peer: not blank
       tags:
       - {key: mq.broker, value: not blank}
       - {key: mq.queue, value: test}

test/plugin/scenarios/kafka-scenario/config/expectedData.yaml

@@ -158,7 +158,7 @@ segmentItems:
       componentId: 41
       isError: false
       spanType: Entry
-      peer: ''
+      peer: kafka-server:9092
       tags:
       - {key: mq.broker, value: 'kafka-server:9092'}
       - {key: mq.topic, value: test}
@@ -194,7 +194,7 @@ segmentItems:
         componentId: 41
         isError: false
         spanType: Entry
-        peer: ''
+        peer: kafka-server:9092
         tags:
           - {key: mq.broker, value: 'kafka-server:9092'}
           - {key: mq.topic, value: test.}
@@ -241,7 +241,7 @@ segmentItems:
         componentId: 41
         isError: false
         spanType: Entry
-        peer: ''
+        peer: kafka-server:9092
         tags:
           - { key: mq.broker, value: 'kafka-server:9092' }
           - { key: mq.topic, value: assign }