Polish up kafka,nats,activemq,rabbitmq plugins to set peer at comsume…

…r side (#377)
apache · Nov 6, 2022 · d4d21ca · d4d21ca
1 parent f17a4ea
commit d4d21ca
Show file tree

Hide file tree

Showing 15 changed files with 29 additions and 17 deletions.
diff --git a/CHANGES.md b/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
 

diff --git a/...ain/java/org/apache/skywalking/apm/plugin/activemq/MessageConsumerDequeueInterceptor.java b/...ain/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();

diff --git a/...plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaConsumerInterceptor.java b/...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();

diff --git a/...n/src/main/java/org/apache/skywalking/apm/plugin/nats/client/DeliverReplyInterceptor.java b/...n/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

diff --git a/...2.15.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/nats/client/NatsCommons.java b/...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) {

diff --git a/...ain/java/org/apache/skywalking/apm/plugin/nats/client/SubscriptionNextMsgInterceptor.java b/...ain/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;

diff --git a/...mq-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rabbitmq/TracerConsumer.java b/...mq-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()) {

diff --git a/docs/en/setup/service-agent/java-agent/Java-Plugin-Development-Guide.md b/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
 

diff --git a/test/plugin/scenarios/activemq-scenario/config/expectedData.yaml b/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}

diff --git a/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml b/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 }

diff --git a/test/plugin/scenarios/nats-2.14.x-2.15.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/nats-2.14.x-2.15.x-scenario/config/expectedData.yaml
@@ -430,7 +430,7 @@ segmentItems:
             componentId: 132
             isError: false
             spanType: Entry
-            peer: ''
+            peer: nats://nats-server:4222
             skipAnalysis: false
             tags:
               - {key: transmission.latency, value: ge 0}
@@ -514,7 +514,7 @@ segmentItems:
             componentId: 132
             isError: false
             spanType: Entry
-            peer: ''
+            peer: nats://nats-server:4222
             skipAnalysis: false
             tags:
               - {key: transmission.latency, value: ge 0}
@@ -539,7 +539,7 @@ segmentItems:
             componentId: 132
             isError: false
             spanType: Entry
-            peer: ''
+            peer: nats://nats-server:4222
             skipAnalysis: false
             tags:
               - {key: transmission.latency, value: ge 0}
@@ -602,7 +602,7 @@ segmentItems:
             componentId: 132
             isError: false
             spanType: Entry
-            peer: ''
+            peer: nats://nats-server:4222
             skipAnalysis: false
             tags:
               - {key: transmission.latency, value: ge 0}
@@ -696,7 +696,7 @@ segmentItems:
             componentId: 132
             isError: false
             spanType: Entry
-            peer: ''
+            peer: nats://nats-server:4222
             skipAnalysis: false
             tags:
               - {key: transmission.latency, value: ge 0}

diff --git a/test/plugin/scenarios/rabbitmq-scenario/config/expectedData.yaml b/test/plugin/scenarios/rabbitmq-scenario/config/expectedData.yaml
@@ -28,7 +28,7 @@ segmentItems:
       componentId: 53
       isError: false
       spanType: Entry
-      peer: ''
+      peer: not blank
       tags:
       - {key: mq.broker, value: not blank}
       - {key: mq.topic, value: ''}
@@ -50,7 +50,7 @@ segmentItems:
       componentId: 52
       isError: false
       spanType: Exit
-      peer: not null
+      peer: not blank
       tags:
       - {key: mq.broker, value: not blank}
       - {key: mq.queue, value: test}

diff --git a/test/plugin/scenarios/spring-kafka-1.3.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/spring-kafka-1.3.x-scenario/config/expectedData.yaml
@@ -96,7 +96,7 @@ segmentItems:
             componentId: 41
             isError: false
             spanType: Entry
-            peer: ''
+            peer: kafka-server:9092
             skipAnalysis: false
             tags:
               - {key: mq.broker, value: 'kafka-server:9092'}

diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/config/expectedData.yaml
@@ -96,7 +96,7 @@ segmentItems:
             componentId: 41
             isError: false
             spanType: Entry
-            peer: ''
+            peer: kafka-server:9092
             skipAnalysis: false
             tags:
               - {key: mq.broker, value: 'kafka-server:9092'}

diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml
@@ -110,7 +110,7 @@ segmentItems:
             componentId: 41
             isError: false
             spanType: Entry
-            peer: ''
+            peer: kafka-server:9092
             skipAnalysis: false
             tags:
               - {key: mq.broker, value: 'kafka-server:9092'}