diff --git a/apm-network/src/main/java/org/skywalking/apm/network/trace/component/ComponentsDefine.java b/apm-network/src/main/java/org/skywalking/apm/network/trace/component/ComponentsDefine.java index 0151346e8..8d411d847 100644 --- a/apm-network/src/main/java/org/skywalking/apm/network/trace/component/ComponentsDefine.java +++ b/apm-network/src/main/java/org/skywalking/apm/network/trace/component/ComponentsDefine.java @@ -71,6 +71,8 @@ public class ComponentsDefine { public static final OfficialComponent GRPC = new OfficialComponent(23, "GRPC"); + public static final OfficialComponent ELASTIC_JOB = new OfficialComponent(24, "ElasticJob"); + private static ComponentsDefine instance = new ComponentsDefine(); private String[] components; @@ -80,7 +82,7 @@ public class ComponentsDefine { } public ComponentsDefine() { - components = new String[24]; + components = new String[25]; addComponent(TOMCAT); addComponent(HTTPCLIENT); addComponent(DUBBO); @@ -104,6 +106,7 @@ public class ComponentsDefine { addComponent(SHARDING_JDBC); addComponent(POSTGRESQL); addComponent(GRPC); + addComponent(ELASTIC_JOB); } private void addComponent(OfficialComponent component) { diff --git a/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/pom.xml new file mode 100644 index 000000000..929d4ff0b --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/pom.xml @@ -0,0 +1,54 @@ + + + + apm-sdk-plugin + org.skywalking + 3.3.0-2017 + + 4.0.0 + + apm-elastic-job-2.x-plugin + jar + + elastic-job-2.x-plugin + http://maven.apache.org + + + UTF-8 + + + + + com.dangdang + elastic-job-common-core + [2.0.0,3.0.0) + provided + + + + + + + org.apache.maven.plugins + maven-deploy-plugin + + + + org.apache.maven.plugins + maven-source-plugin + + + + attach-sources + + jar + + + + + + + + diff --git a/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/java/org/skywalking/apm/plugin/esjob/JobExecutorInterceptor.java b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/java/org/skywalking/apm/plugin/esjob/JobExecutorInterceptor.java new file mode 100644 index 000000000..735dbd56d --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/java/org/skywalking/apm/plugin/esjob/JobExecutorInterceptor.java @@ -0,0 +1,65 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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 repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.plugin.esjob; + +import com.dangdang.ddframe.job.api.ShardingContext; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import com.google.common.base.Strings; +import java.lang.reflect.Method; +import org.skywalking.apm.agent.core.context.ContextManager; +import org.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; +import org.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; +import org.skywalking.apm.network.trace.component.ComponentsDefine; + +/** + * {@link JobExecutorInterceptor} enhances {@link com.dangdang.ddframe.job.executor.AbstractElasticJobExecutor#process(ShardingContext)} + * ,creating a local span that records job execution. + * + * @author gaohongtao + */ +public class JobExecutorInterceptor implements InstanceMethodsAroundInterceptor { + @Override + public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, + MethodInterceptResult result) throws Throwable { + ShardingContexts shardingContexts = (ShardingContexts)allArguments[0]; + Integer item = (Integer)allArguments[1]; + ShardingContext shardingContext = new ShardingContext(shardingContexts, item); + String operateName = shardingContext.getJobName(); + if (!Strings.isNullOrEmpty(shardingContext.getShardingParameter())) { + operateName += "-" + shardingContext.getShardingParameter(); + } + AbstractSpan span = ContextManager.createLocalSpan(operateName); + span.setComponent(ComponentsDefine.ELASTIC_JOB); + span.tag("sharding_context", shardingContext.toString()); + } + + @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); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/java/org/skywalking/apm/plugin/esjob/define/JobExecutorInstrumentation.java b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/java/org/skywalking/apm/plugin/esjob/define/JobExecutorInstrumentation.java new file mode 100644 index 000000000..eb29a9e3e --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/java/org/skywalking/apm/plugin/esjob/define/JobExecutorInstrumentation.java @@ -0,0 +1,68 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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 repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.plugin.esjob.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint; +import org.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; +import org.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine; +import org.skywalking.apm.agent.core.plugin.match.ClassMatch; + +import static org.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType; +import static org.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * {@link JobExecutorInstrumentation} presents that skywalking intercepts {@link com.dangdang.ddframe.job.executor.AbstractElasticJobExecutor}. + * + * @author gaohongtao + */ +public class JobExecutorInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { + + private static final String ENHANCE_CLASS = "com.dangdang.ddframe.job.executor.AbstractElasticJobExecutor"; + + private static final String JOB_EXECUTOR_INTERCEPTOR_CLASS = "org.skywalking.apm.plugin.esjob.JobExecutorInterceptor"; + + @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { + return new ConstructorInterceptPoint[0]; + } + + @Override protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[]{ + new InstanceMethodsInterceptPoint() { + + @Override public ElementMatcher getMethodsMatcher() { + return takesArgumentWithType(2, "com.dangdang.ddframe.job.event.type.JobExecutionEvent"); + } + + @Override public String getMethodsInterceptor() { + return JOB_EXECUTOR_INTERCEPTOR_CLASS; + } + + @Override public boolean isOverrideArgs() { + return false; + } + } + }; + } + + @Override protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS); + } +} diff --git a/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/resources/skywalking-plugin.def new file mode 100644 index 000000000..7e8a4d707 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/main/resources/skywalking-plugin.def @@ -0,0 +1 @@ +elastic-job-2.x=org.skywalking.apm.plugin.esjob.define.JobExecutorInstrumentation \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/test/java/org/skywalking/apm/plugin/esjob/JobExecutorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/test/java/org/skywalking/apm/plugin/esjob/JobExecutorInterceptorTest.java new file mode 100644 index 000000000..6e946a709 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/elastic-job-2.x-plugin/src/test/java/org/skywalking/apm/plugin/esjob/JobExecutorInterceptorTest.java @@ -0,0 +1,107 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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 repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.plugin.esjob; + +import com.dangdang.ddframe.job.executor.ShardingContexts; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +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 org.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.skywalking.apm.agent.core.context.trace.TraceSegment; +import org.skywalking.apm.agent.test.helper.SegmentHelper; +import org.skywalking.apm.agent.test.tools.AgentServiceRule; +import org.skywalking.apm.agent.test.tools.SegmentStorage; +import org.skywalking.apm.agent.test.tools.SegmentStoragePoint; +import org.skywalking.apm.agent.test.tools.TracingSegmentRunner; + +import static org.hamcrest.core.Is.is; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +@RunWith(PowerMockRunner.class) +@PowerMockRunnerDelegate(TracingSegmentRunner.class) +public class JobExecutorInterceptorTest { + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + + private JobExecutorInterceptor jobExecutorInterceptor; + + @Before + public void setUp() throws SQLException { + jobExecutorInterceptor = new JobExecutorInterceptor(); + } + + @Test + public void assertSuccess() throws Throwable { + jobExecutorInterceptor.beforeMethod(null, null, new Object[]{mockShardingContext("fooJob", 1), 1}, null, null); + jobExecutorInterceptor.afterMethod(null, null, null, null, null); + TraceSegment segment = segmentStorage.getTraceSegments().get(0); + List spans = SegmentHelper.getSpans(segment); + assertNotNull(spans); + assertThat(spans.size(), is(1)); + assertThat(spans.get(0).transform().getOperationName(), is("fooJob-test")); + assertThat(spans.get(0).transform().getComponentId(), is(24)); + assertThat(spans.get(0).transform().getTags(0).getKey(), is("sharding_context")); + assertThat(spans.get(0).transform().getTags(0).getValue(), is("ShardingContext(jobName=fooJob, taskId=fooJob1, shardingTotalCount=2, jobParameter=, shardingItem=1, shardingParameter=test)")); + } + + @Test + public void assertSuccessWithoutSharding() throws Throwable { + jobExecutorInterceptor.beforeMethod(null, null, new Object[]{mockShardingContext("fooJob", 0), 0}, null, null); + jobExecutorInterceptor.afterMethod(null, null, null, null, null); + TraceSegment segment = segmentStorage.getTraceSegments().get(0); + List spans = SegmentHelper.getSpans(segment); + assertNotNull(spans); + assertThat(spans.size(), is(1)); + assertThat(spans.get(0).transform().getOperationName(), is("fooJob")); + assertThat(spans.get(0).transform().getTags(0).getValue(), is("ShardingContext(jobName=fooJob, taskId=fooJob0, shardingTotalCount=1, jobParameter=, shardingItem=0, shardingParameter=null)")); + } + + @Test + public void assertError() throws Throwable { + jobExecutorInterceptor.beforeMethod(null, null, new Object[]{mockShardingContext("fooJob", 0), 0}, null, null); + jobExecutorInterceptor.handleMethodException(null, null, null, null, new Exception("fooError")); + jobExecutorInterceptor.afterMethod(null, null, null, null, null); + TraceSegment segment = segmentStorage.getTraceSegments().get(0); + List spans = SegmentHelper.getSpans(segment); + assertNotNull(spans); + assertThat(spans.size(), is(1)); + assertThat(spans.get(0).transform().getIsError(), is(true)); + assertThat(spans.get(0).transform().getLogs(0).getDataList().size(), is(4)); + } + + private ShardingContexts mockShardingContext(String jobName, int shardingItem) { + Map shardingMap = new HashMap(1); + if (shardingItem >= 1) { + shardingMap.put(shardingItem, "test"); + } + return new ShardingContexts(jobName + shardingItem, jobName, shardingItem + 1, "", shardingMap); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/pom.xml index 32bb38fba..edfd655fb 100644 --- a/apm-sniffer/apm-sdk-plugin/pom.xml +++ b/apm-sniffer/apm-sdk-plugin/pom.xml @@ -52,6 +52,7 @@ h2-1.x-plugin postgresql-8.x-plugin oracle-10.x-plugin + elastic-job-2.x-plugin mongodb-2.x-plugin pom