diff --git a/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java index b23f30cf3..220b1062c 100644 --- a/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java +++ b/apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java @@ -78,6 +78,8 @@ public class ComponentsDefine { public static final OfficialComponent HTTP_ASYNC_CLIENT = new OfficialComponent(26, "httpasyncclient"); + public static final OfficialComponent KAFKA = new OfficialComponent(27, "Kafka"); + private static ComponentsDefine INSTANCE = new ComponentsDefine(); private String[] components; @@ -87,7 +89,7 @@ public class ComponentsDefine { } public ComponentsDefine() { - components = new String[27]; + components = new String[28]; addComponent(TOMCAT); addComponent(HTTPCLIENT); addComponent(DUBBO); @@ -114,6 +116,7 @@ public class ComponentsDefine { addComponent(ELASTIC_JOB); addComponent(ROCKET_MQ); addComponent(HTTP_ASYNC_CLIENT); + addComponent(KAFKA); } private void addComponent(OfficialComponent component) { diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/pom.xml new file mode 100644 index 000000000..37699929d --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/pom.xml @@ -0,0 +1,40 @@ + + + + + + apm-sdk-plugin + org.apache.skywalking + 5.0.0-alpha + + 4.0.0 + + apm-kafka-0.11.x-plugin + + + + org.apache.kafka + kafka-clients + 0.11.0.0 + provided + + + \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptor.java new file mode 100644 index 000000000..a0f81a084 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptor.java @@ -0,0 +1,55 @@ +/* + * 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.lang.reflect.Method; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; + +public class CallbackInterceptor implements InstanceMethodsAroundInterceptor { + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + AbstractSpan abstractSpan = ContextManager.createLocalSpan("Producer/Callback"); + + //Get the SnapshotContext + ContextSnapshot contextSnapshot = (ContextSnapshot)objInst.getSkyWalkingDynamicField(); + ContextManager.continued(contextSnapshot); + } + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + Exception exceptions = (Exception)allArguments[1]; + if (exceptions != null) { + ContextManager.activeSpan().errorOccurred().log(exceptions); + } + ContextManager.stopSpan(); + return ret; + } + + @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + ContextManager.activeSpan().errorOccurred().log(t); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptor.java new file mode 100644 index 000000000..fd3bc6c9e --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptor.java @@ -0,0 +1,34 @@ +/* + * 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.kafka.clients.consumer.ConsumerConfig; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor; + +public class ConsumerConstructorInterceptor implements InstanceConstructorInterceptor { + + @Override public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { + ConsumerConfig config = (ConsumerConfig)allArguments[0]; + // set the bootstrap server address + ConsumerEnhanceRequiredInfo requiredInfo = new ConsumerEnhanceRequiredInfo(); + requiredInfo.setBrokerServers(config.getList("bootstrap.servers")); + objInst.setSkyWalkingDynamicField(requiredInfo); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java new file mode 100644 index 000000000..2eb660137 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerEnhanceRequiredInfo.java @@ -0,0 +1,44 @@ +/* + * 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.Collection; +import java.util.List; +import org.apache.skywalking.apm.util.StringUtil; + +public class ConsumerEnhanceRequiredInfo { + private String brokerServers; + private String topics; + + public void setBrokerServers(List brokerServers) { + this.brokerServers = StringUtil.join(',', brokerServers.toArray(new String[0])); + } + + public void setTopics(Collection topics) { + this.topics = StringUtil.join(',', topics.toArray(new String[0])); + } + + public String getBrokerServers() { + return brokerServers; + } + + public String getTopics() { + return topics; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java new file mode 100644 index 000000000..f6804cc2f --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java @@ -0,0 +1,83 @@ +/* + * 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.lang.reflect.Method; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.header.Header; +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; +import org.apache.skywalking.apm.agent.core.context.tag.Tags; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; + +/** + * @autor zhang xin + */ +public class KafkaConsumerInterceptor implements InstanceMethodsAroundInterceptor { + @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()); + } + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + Map>> records = (Map>>)ret; + for (List> consumerRecords : records.values()) { + for (ConsumerRecord record : consumerRecords) { + ContextCarrier contextCarrier = new ContextCarrier(); + + CarrierItem next = contextCarrier.items(); + while (next.hasNext()) { + next = next.next(); + Iterator
iterator = record.headers().headers(next.getHeadKey()).iterator(); + if (iterator.hasNext()) { + next.setHeadValue(new String(iterator.next().value())); + } + } + ContextManager.extract(contextCarrier); + } + } + + ContextManager.stopSpan(); + return ret; + } + + @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + ContextManager.activeSpan().errorOccurred().log(t); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java new file mode 100644 index 000000000..929d071cb --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java @@ -0,0 +1,72 @@ +/* + * 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.lang.reflect.Method; +import org.apache.kafka.clients.producer.ProducerRecord; +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; +import org.apache.skywalking.apm.agent.core.context.tag.Tags; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; + +/** + * @author zhang xin + */ +public class KafkaProducerInterceptor implements InstanceMethodsAroundInterceptor { + @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()); + + //set tags + ProducerRecord record = (ProducerRecord)allArguments[0]; + Tags.MQ_BROKER.set(activeSpan, (String)objInst.getSkyWalkingDynamicField()); + Tags.MQ_TOPIC.set(activeSpan, (String)((EnhancedInstance)record).getSkyWalkingDynamicField()); + + // set headers + CarrierItem next = contextCarrier.items(); + while (next.hasNext()) { + next = next.next(); + record.headers().add(next.getHeadKey(), next.getHeadValue().getBytes()); + } + + EnhancedInstance callbackInstance = (EnhancedInstance)allArguments[1]; + if (callbackInstance != null) { + callbackInstance.setSkyWalkingDynamicField(ContextManager.capture()); + } + } + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + ContextManager.stopSpan(); + return ret; + } + + @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java new file mode 100644 index 000000000..ed267df21 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java @@ -0,0 +1,33 @@ +/* + * 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.kafka.clients.producer.ProducerConfig; +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.util.StringUtil; + +public class ProducerConstructorInterceptor implements InstanceConstructorInterceptor { + + @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]))); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptor.java new file mode 100644 index 000000000..0cb723a43 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptor.java @@ -0,0 +1,30 @@ +/* + * 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.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor; + +public class ProducerRecordConstructorInterceptor implements InstanceConstructorInterceptor { + @Override public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { + String topic = (String)allArguments[0]; + // set the topic + objInst.setSkyWalkingDynamicField(topic); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptor.java new file mode 100644 index 000000000..657684a9f --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptor.java @@ -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.kafka.v11; + +import java.lang.reflect.Method; +import java.util.Collection; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; + +public class SubscribeMethodInterceptor implements InstanceMethodsAroundInterceptor { + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo)objInst.getSkyWalkingDynamicField(); + requiredInfo.setTopics((Collection)allArguments[0]); + + objInst.setSkyWalkingDynamicField(requiredInfo); + } + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + Object ret) throws Throwable { + return ret; + } + + @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, + Class[] argumentsTypes, Throwable t) { + ContextManager.activeSpan().errorOccurred().log(t); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/CallbackInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/CallbackInstrumentation.java new file mode 100644 index 000000000..715166888 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/CallbackInstrumentation.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.plugin.kafka.v11.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierarchyMatch; + +/** + * {@link CallbackInstrumentation} intercept the method onCompletion in the class org.apache.kafka.clients.producer.Callback. + * + * @author zhangxin + */ +public class CallbackInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + public static final String ENHANCE_CLASS = "org.apache.kafka.clients.producer.Callback"; + public static final String ENHANCE_METHOD = "onCompletion"; + public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.CallbackInterceptor"; + + @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[0]; + } + + @Override protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[] { + new InstanceMethodsInterceptPoint() { + @Override public ElementMatcher getMethodsMatcher() { + return named(ENHANCE_METHOD); + } + + @Override public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override public boolean isOverrideArgs() { + return false; + } + } + }; + } + + @Override protected ClassMatch enhanceClass() { + return byHierarchyMatch(new String[] {ENHANCE_CLASS}); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaConsumerInstrumentation.java new file mode 100644 index 000000000..81b9bbe87 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaConsumerInstrumentation.java @@ -0,0 +1,104 @@ +/* + * 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.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * {@link KafkaProducerInstrumentation} intercept the method send in the class + * org.apache.kafka.clients.producer.KafkaProducer. Here is the intercept process steps. + * + * + * 
+ *  1. Record the topic when the client call subscribed
+ *  3. Create the entry span when the client call the method pollOnce.
+ *  4. Inject all the Trace Context by iterate all ConsumerRecord
+ *  5. Stop the entry span when end the pollOnce method.
+ *
+ * + * @author zhang xin + */ +public class KafkaConsumerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + public static final String CONSTRUCTOR_INTERCEPT_FLAG = "org.apache.kafka.clients.consumer.ConsumerConfig"; + public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.ConsumerConstructorInterceptor"; + public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.KafkaConsumerInterceptor"; + public static final String ENHANCE_METHOD = "pollOnce"; + public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; + public static final String SUBSCRIBE_METHOD = "subscribe"; + public static final String SUBSCRIBE_INTERCEPT_FLAG = "org.apache.kafka.clients.consumer.ConsumerRebalanceListener"; + + @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_FLAG); + } + + @Override public String getConstructorInterceptor() { + return CONSTRUCTOR_INTERCEPTOR_CLASS; + } + } + }; + } + + @Override protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[] { + new InstanceMethodsInterceptPoint() { + @Override public ElementMatcher getMethodsMatcher() { + return named(ENHANCE_METHOD); + } + + @Override public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD).and(takesArgumentWithType(1, SUBSCRIBE_INTERCEPT_FLAG)); + } + + @Override public String getMethodsInterceptor() { + return "org.apache.skywalking.apm.plugin.kafka.v11.SubscribeMethodInterceptor"; + } + + @Override public boolean isOverrideArgs() { + return false; + } + } + }; + } + + @Override protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaProducerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaProducerInstrumentation.java new file mode 100644 index 000000000..cd3ff6041 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaProducerInstrumentation.java @@ -0,0 +1,91 @@ +/* + * 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.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * {@link KafkaProducerInstrumentation} intercept the method send in the class + * org.apache.kafka.clients.producer.KafkaProducer. Here is the intercept process steps. + * + * + * 
+ *  1. Record the broker address when the client create the org.apache.kafka.clients.producer.KafkaProducer
+ * instance
+ *  2. Fetch the topic name from org.apache.kafka.clients.producer.ProducerRecord when the client call
+ * send method.
+ *  3. Create the exit span when the client call send method
+ *  4. Set the Context into the org.apache.kafka.clients.producer.ProducerRecord#headers
+ *  5. Stop the exit span when end the send method.
+ *
+ * + * @author zhang xin + */ +public class KafkaProducerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.KafkaProducerInterceptor"; + public static final String ENHANCE_CLASS = "org.apache.kafka.clients.producer.KafkaProducer"; + public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.ProducerConstructorInterceptor"; + public static final String CONSTRUCTOR_INTERCEPTOR_FLAG = "org.apache.kafka.clients.producer.ProducerConfig"; + + @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPTOR_FLAG); + } + + @Override public String getConstructorInterceptor() { + return CONSTRUCTOR_INTERCEPTOR_CLASS; + } + } + }; + } + + @Override protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[] { + new InstanceMethodsInterceptPoint() { + @Override public ElementMatcher getMethodsMatcher() { + return named("doSend"); + } + + @Override public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override public boolean isOverrideArgs() { + return false; + } + } + }; + } + + @Override protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/ProducerRecordInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/ProducerRecordInstrumentation.java new file mode 100644 index 000000000..abcee1a0a --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/ProducerRecordInstrumentation.java @@ -0,0 +1,64 @@ +/* + * 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.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * {@link ProducerRecordInstrumentation} intercept the constructor in the class org.apache.kafka.clients.producer.ProducerRecord + * for record the topic name and propagate the Context of trace. + * + * @author zhang xin + * @see org.apache.skywalking.apm.plugin.kafka.v11.define.KafkaProducerInstrumentation + */ +public class ProducerRecordInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.ProducerRecordConstructorInterceptor"; + public static final String ENHANCE_CLASS = "org.apache.kafka.clients.producer.ProducerRecord"; + + @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override public ElementMatcher getConstructorMatcher() { + return takesArguments(6); + } + + @Override public String getConstructorInterceptor() { + return CONSTRUCTOR_INTERCEPTOR_CLASS; + } + } + }; + } + + @Override protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[0]; + } + + @Override protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/resources/skywalking-plugin.def new file mode 100644 index 000000000..5f3439cbf --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/resources/skywalking-plugin.def @@ -0,0 +1,4 @@ +kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v11.define.CallbackInstrumentation +kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v11.define.KafkaConsumerInstrumentation +kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v11.define.KafkaProducerInstrumentation +kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v11.define.ProducerRecordInstrumentation diff --git a/apm-sniffer/apm-sdk-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/pom.xml index 4b81d1d8c..fa8d0b0ae 100644 --- a/apm-sniffer/apm-sdk-plugin/pom.xml +++ b/apm-sniffer/apm-sdk-plugin/pom.xml @@ -53,6 +53,7 @@ elastic-job-2.x-plugin mongodb-2.x-plugin httpasyncclient-4.x-plugin + kafka-0.11.x-plugin pom