This is an automated email from the ASF dual-hosted git repository. zhangxin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/incubator-skywalking.git
The following commit(s) were added to refs/heads/master by this push: new addf510 Support canal plugin (#2035) addf510 is described below commit addf5101e006f73467b2c43c3e477c4c8af96cb6 Author: XiaoFu <248628...@qq.com> AuthorDate: Thu Dec 20 21:07:31 2018 +0800 Support canal plugin (#2035) * fix add Canal plugin * fix add UnitTest * fix format pom.xml * Delete CanalApplication.java * fix support cluster url * fix add Licensed * fix bug * fix --- .../network/trace/component/ComponentsDefine.java | 5 +- .../apm-sdk-plugin/canal-1.x-plugin/pom.xml | 45 ++++++++ .../plugin/canal/CanalConstructorInterceptor.java | 44 ++++++++ .../apm/plugin/canal/CanalEnhanceInfo.java | 51 +++++++++ .../apm/plugin/canal/CanalInterceptor.java | 85 +++++++++++++++ .../canal/ClusterNodeConstructInterceptor.java | 61 +++++++++++ .../plugin/canal/define/CanalInstrumentation.java | 81 +++++++++++++++ .../canal/define/ClusterNodeInstrumentation.java | 63 +++++++++++ .../src/main/resources/skywalking-plugin.def | 18 ++++ .../apm/plugin/canal/CanalInterceptorTest.java | 115 +++++++++++++++++++++ apm-sniffer/apm-sdk-plugin/pom.xml | 3 +- docker/config/component-libraries.yml | 3 + .../src/main/resources/component-libraries.yml | 4 + 13 files changed, 576 insertions(+), 2 deletions(-) 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 73da6a8..fd155d1 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 @@ -104,6 +104,8 @@ public class ComponentsDefine { public static final OfficialComponent RABBITMQ_CONSUMER = new OfficialComponent(53,"rabbitmq-consumer"); + public static final OfficialComponent CANAL = new OfficialComponent(54,"Canal"); + private static ComponentsDefine INSTANCE = new ComponentsDefine(); private String[] components; @@ -113,7 +115,7 @@ public class ComponentsDefine { } public ComponentsDefine() { - components = new String[54]; + components = new String[55]; addComponent(TOMCAT); addComponent(HTTPCLIENT); addComponent(DUBBO); @@ -152,6 +154,7 @@ public class ComponentsDefine { addComponent(UNDERTOW); addComponent(RABBITMQ_PRODUCER); addComponent(RABBITMQ_CONSUMER); + addComponent(CANAL); } private void addComponent(OfficialComponent component) { diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/pom.xml new file mode 100644 index 0000000..3391c0a --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/pom.xml @@ -0,0 +1,45 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ 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. + ~ + --> + +<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + <parent> + <artifactId>apm-sdk-plugin</artifactId> + <groupId>org.apache.skywalking</groupId> + <version>6.0.0-beta-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>apm-canal-1.x-plugin</artifactId> + <name>canal-1.x-plugin</name> + <packaging>jar</packaging> + + <properties> + <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> + <canal-client.version>1.1.2</canal-client.version> + </properties> + + <dependencies> + <dependency> + <groupId>com.alibaba.otter</groupId> + <artifactId>canal.client</artifactId> + <version>${canal-client.version}</version> + <scope>provided</scope> + </dependency> + </dependencies> +</project> diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalConstructorInterceptor.java new file mode 100644 index 0000000..b803ca0 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalConstructorInterceptor.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.canal; + +import java.net.InetSocketAddress; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor; + + +/** + * @author withlin + */ +public class CanalConstructorInterceptor implements InstanceConstructorInterceptor { + @Override + public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { + InetSocketAddress address = (InetSocketAddress) allArguments[0]; + String destination = allArguments[3].toString(); + CanalEnhanceInfo canalEnhanceInfo = new CanalEnhanceInfo(); + if (address != null) { + String url = address.getAddress().toString() + ":" + address.getPort(); + canalEnhanceInfo.setUrl(url.replace('/',' ')); + } + canalEnhanceInfo.setDestination(destination); + objInst.setSkyWalkingDynamicField(canalEnhanceInfo); + + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalEnhanceInfo.java b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalEnhanceInfo.java new file mode 100644 index 0000000..642ba38 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalEnhanceInfo.java @@ -0,0 +1,51 @@ +/* + * 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.canal; + +/** + * @author withlin + */ +public class CanalEnhanceInfo { + + public String getUrl() { + return url; + } + + public CanalEnhanceInfo setUrl(String url) { + this.url = url; + return this; + } + + public String getDestination() { + return destination; + } + + public CanalEnhanceInfo setDestination(String destination) { + this.destination = destination; + return this; + } + /** + * canal-server address + */ + private String url; + /** + * canal destination + */ + private String destination; +} diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalInterceptor.java b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalInterceptor.java new file mode 100644 index 0000000..3785826 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/CanalInterceptor.java @@ -0,0 +1,85 @@ +/* + * 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.canal; + +import com.alibaba.otter.canal.client.impl.SimpleCanalConnector; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +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; +import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; + +import java.lang.reflect.Method; +import java.net.InetSocketAddress; +import java.util.List; + +/** + * @author withlin + */ +public class CanalInterceptor implements InstanceMethodsAroundInterceptor { + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, + Class<?>[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + CanalEnhanceInfo canalEnhanceInfo = (CanalEnhanceInfo)objInst.getSkyWalkingDynamicField(); + SimpleCanalConnector connector = (SimpleCanalConnector) objInst; + + String url = canalEnhanceInfo.getUrl(); + if (url == "" || url == null) { + InetSocketAddress address = (InetSocketAddress)connector.getNextAddress(); + String runningAddress = address.getAddress().toString() + ":" + address.getPort(); + runningAddress = runningAddress.replace('/',' '); + url = runningAddress; + List<InetSocketAddress> socketAddressList = (List<InetSocketAddress>)ContextManager.getRuntimeContext().get("currentAddress"); + if (socketAddressList != null && socketAddressList.size() > 0) { + for (InetSocketAddress socketAddress : socketAddressList) { + String currentAddress = socketAddress.getAddress().toString() + ":" + socketAddress.getPort(); + currentAddress = currentAddress.replace('/',' '); + if (!currentAddress.equals(runningAddress)) { + url = url + "," + currentAddress; + } + } + } + } + String batchSize = allArguments[0].toString(); + String destination = canalEnhanceInfo.getDestination(); + AbstractSpan activeSpan = ContextManager.createExitSpan("Canal/" + destination,url).start(System.currentTimeMillis()); + activeSpan.setComponent(ComponentsDefine.CANAL); + activeSpan.tag("batchSize",batchSize); + activeSpan.tag("destination",destination); + + } + + @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) { + ContextManager.activeSpan().errorOccurred().log(t); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/ClusterNodeConstructInterceptor.java b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/ClusterNodeConstructInterceptor.java new file mode 100644 index 0000000..5d58492 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/ClusterNodeConstructInterceptor.java @@ -0,0 +1,61 @@ +/* + * 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.canal; + +import com.alibaba.otter.canal.client.impl.ClusterNodeAccessStrategy; +import com.alibaba.otter.canal.common.zookeeper.ZkClientx; +import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils; +import org.apache.commons.lang.StringUtils; +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.InstanceConstructorInterceptor; + +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.List; + + +/** + * @author withlin + */ +public class ClusterNodeConstructInterceptor implements InstanceConstructorInterceptor { + @Override + public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { + + String clusterPath = ZookeeperPathUtils.getDestinationClusterRoot(allArguments[0].toString()); + ZkClientx zkClientx = ((ClusterNodeAccessStrategy) objInst).getZkClient(); + ContextManager.getRuntimeContext().put("currentAddress",getCurrentAddress(zkClientx.getChildren(clusterPath))); + + } + + private List<InetSocketAddress> getCurrentAddress(List<String> currentChilds) { + List<InetSocketAddress> addresses = new ArrayList<InetSocketAddress>(); + for (String address : currentChilds) { + String[] strs = StringUtils.split(address, ":"); + if (strs != null && strs.length == 2) { + addresses.add(new InetSocketAddress(strs[0], Integer.valueOf(strs[1]))); + } + } + + return addresses; + + } +} + + diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/define/CanalInstrumentation.java b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/define/CanalInstrumentation.java new file mode 100644 index 0000000..2c8308c --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/define/CanalInstrumentation.java @@ -0,0 +1,81 @@ +/* + * 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.canal.define; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; + +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 org.apache.skywalking.apm.agent.core.plugin.match.MultiClassNameMatch; + + +/** + * @author withlin + */ +public class CanalInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.canal.CanalInterceptor"; + public static final String ENHANCE_CLASS = "com.alibaba.otter.canal.client.impl.SimpleCanalConnector"; + public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.canal.CanalConstructorInterceptor"; + public static final String ENHANCE_METHOD_DISPATCH = "getWithoutAck"; + + @Override + protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override public ElementMatcher<MethodDescription> getConstructorMatcher() { + return takesArgument(4, int.class); + } + + @Override public String getConstructorInterceptor() { + return CONSTRUCTOR_INTERCEPTOR_CLASS; + } + } + }; + } + + @Override + protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[] { + new InstanceMethodsInterceptPoint() { + @Override public ElementMatcher<MethodDescription> getMethodsMatcher() { + return named(ENHANCE_METHOD_DISPATCH); + } + + + @Override public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } + + @Override public boolean isOverrideArgs() { + return false; + } + } + }; + } + + @Override + protected ClassMatch enhanceClass() { + return MultiClassNameMatch.byMultiClassMatch(ENHANCE_CLASS); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/define/ClusterNodeInstrumentation.java b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/define/ClusterNodeInstrumentation.java new file mode 100644 index 0000000..21d05d2 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/canal/define/ClusterNodeInstrumentation.java @@ -0,0 +1,63 @@ +/* + * 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.canal.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 org.apache.skywalking.apm.agent.core.plugin.match.MultiClassNameMatch; + +import static net.bytebuddy.matcher.ElementMatchers.any; + + +/** + * @author withlin + */ +public class ClusterNodeInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + public static final String ENHANCE_CLASS = "com.alibaba.otter.canal.client.impl.ClusterNodeAccessStrategy"; + public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.canal.ClusterNodeConstructInterceptor"; + + @Override + protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override public ElementMatcher<MethodDescription> getConstructorMatcher() { + return any(); + } + + @Override public String getConstructorInterceptor() { + return CONSTRUCTOR_INTERCEPTOR_CLASS; + } + } + }; + } + + @Override + protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[0]; + } + + @Override + protected ClassMatch enhanceClass() { + return MultiClassNameMatch.byMultiClassMatch(ENHANCE_CLASS); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/resources/skywalking-plugin.def new file mode 100644 index 0000000..f01d808 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/main/resources/skywalking-plugin.def @@ -0,0 +1,18 @@ +# 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. + +canal-1.x=org.apache.skywalking.apm.plugin.canal.define.CanalInstrumentation +canal-1.x=org.apache.skywalking.apm.plugin.canal.define.ClusterNodeInstrumentation \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/canal/CanalInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/canal/CanalInterceptorTest.java new file mode 100644 index 0000000..c9c6fc3 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/canal-1.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/canal/CanalInterceptorTest.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.plugin.canal; + +import com.alibaba.otter.canal.client.impl.SimpleCanalConnector; +import com.alibaba.otter.canal.common.utils.AddressUtils; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.test.helper.SegmentHelper; +import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule; +import org.apache.skywalking.apm.agent.test.tools.SegmentStorage; +import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint; +import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.modules.junit4.PowerMockRunner; +import org.powermock.modules.junit4.PowerMockRunnerDelegate; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.List; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; + +@RunWith(PowerMockRunner.class) +@PowerMockRunnerDelegate(TracingSegmentRunner.class) +public class CanalInterceptorTest { + + private CanalInterceptor canalInterceptor; + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + + private Object[] arguments; + + private Class[] argumentType; + + + private class CanalConnector extends SimpleCanalConnector implements EnhancedInstance { + + public CanalConnector(SocketAddress address, String username, String password, String destination) { + super(address, username, password, destination); + } + + @Override + public Object getSkyWalkingDynamicField() { + return null; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + + } + } + + + private EnhancedInstance enhancedInstance = new CanalConnector(new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example", "", "") { + @Override + public Object getSkyWalkingDynamicField() { + CanalEnhanceInfo canalEnhanceInfo = new CanalEnhanceInfo(); + canalEnhanceInfo.setUrl("localhost:11111"); + canalEnhanceInfo.setDestination("example"); + return canalEnhanceInfo; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + } + + + }; + + @Before + public void setUp() { + canalInterceptor = new CanalInterceptor(); + arguments = new Object[] {100}; + } + + @Test + public void testSendMessage() throws Throwable { + canalInterceptor.beforeMethod(enhancedInstance, null, arguments, null, null); + canalInterceptor.afterMethod(enhancedInstance, null, arguments, null, null); + + List<TraceSegment> traceSegmentList = segmentStorage.getTraceSegments(); + assertThat(traceSegmentList.size(), is(1)); + + TraceSegment segment = traceSegmentList.get(0); + List<AbstractTracingSpan> spans = SegmentHelper.getSpans(segment); + assertThat(spans.size(), is(1)); + } + +} diff --git a/apm-sniffer/apm-sdk-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/pom.xml index 49d687c..e3f07bd 100644 --- a/apm-sniffer/apm-sdk-plugin/pom.xml +++ b/apm-sniffer/apm-sdk-plugin/pom.xml @@ -60,7 +60,8 @@ <module>activemq-5.x-plugin</module> <module>elasticsearch-5.x-plugin</module> <module>undertow-plugins</module> - <module>rabbitmq-5.x-plugin</module> + <module>rabbitmq-5.x-plugin</module> + <module>canal-1.x-plugin</module> </modules> <packaging>pom</packaging> diff --git a/docker/config/component-libraries.yml b/docker/config/component-libraries.yml index caa6bfb..0c1c670 100644 --- a/docker/config/component-libraries.yml +++ b/docker/config/component-libraries.yml @@ -186,6 +186,9 @@ rabbitmq-producer: rabbitmq-consumer: id: 53 languages: Java +Canal: + id: 54 + languages: Java # .NET/.NET Core components # [3000, 4000) for C#/.NET only diff --git a/oap-server/server-starter/src/main/resources/component-libraries.yml b/oap-server/server-starter/src/main/resources/component-libraries.yml index caa6bfb..ba03e96 100644 --- a/oap-server/server-starter/src/main/resources/component-libraries.yml +++ b/oap-server/server-starter/src/main/resources/component-libraries.yml @@ -186,6 +186,10 @@ rabbitmq-producer: rabbitmq-consumer: id: 53 languages: Java +Canal: + id: 54 + languages: Java + # .NET/.NET Core components # [3000, 4000) for C#/.NET only