From 049dc2fae8b83a6586821836a740491a4074128b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E6=99=9F=20Wu=20Sheng?= Date: Sun, 1 Apr 2018 11:16:01 +0800 Subject: [PATCH] Support collector service instrumentation (#730) * Collector instrument agent. * Make the agent.jar output to the /agent folder in collector package. * Try to add metric annotation. * Change the instrument ways. * Reformat. * Finish the agent codes. Wait for @peng-yongsheng 's metric requirements. * Debug new instrument. * Fix a detect bug. * Finish the instrument * Fix compile issue. --- .../parser/provider/parser/SegmentParse.java | 5 + .../model/base/LocalAsyncWorkerRef.java | 6 +- apm-collector/apm-collector-boot/pom.xml | 8 + .../src/main/assembly/assembly.xml | 7 + .../annotations/trace/BatchParameter.java | 32 ++++ .../trace/GraphComputingMetric.java | 38 ++++ .../annotations/trace/TracedGraphElement.java | 39 ++++ .../apm-collector-instrument/pom.xml | 41 +++++ .../instrument/CollectorInstrumentAgent.java | 90 +++++++++ .../collector/instrument/MetricCollector.java | 172 ------------------ .../apm/collector/instrument/MetricTree.java | 160 ++++++++++++++++ .../{TracedService.java => ReportWriter.java} | 13 +- .../instrument/ServiceInstrumentation.java | 73 -------- .../collector/instrument/ServiceMetric.java | 61 +++++++ .../instrument/ServiceMetricBatchRecord.java | 51 ++++++ .../instrument/ServiceMetricRecord.java | 58 ++++++ .../instrument/ServiceMetricTracing.java | 27 ++- 17 files changed, 621 insertions(+), 260 deletions(-) create mode 100644 apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/BatchParameter.java create mode 100644 apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/GraphComputingMetric.java create mode 100644 apm-collector/apm-collector-core/src/test/java/org/apache/skywalking/apm/collector/core/annotations/trace/TracedGraphElement.java create mode 100644 apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/CollectorInstrumentAgent.java delete mode 100644 apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricCollector.java create mode 100644 apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricTree.java rename apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/{TracedService.java => ReportWriter.java} (76%) delete mode 100644 apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceInstrumentation.java create mode 100644 apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetric.java create mode 100644 apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricBatchRecord.java create mode 100644 apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricRecord.java diff --git a/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/SegmentParse.java b/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/SegmentParse.java index 060bb88a4..b8094cb98 100644 --- a/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/SegmentParse.java +++ b/apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/SegmentParse.java @@ -36,6 +36,7 @@ import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.pars import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser.standardization.SegmentStandardization; import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser.standardization.SpanIdExchanger; import org.apache.skywalking.apm.collector.core.UnexpectedException; +import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric; import org.apache.skywalking.apm.collector.core.graph.Graph; import org.apache.skywalking.apm.collector.core.graph.GraphManager; import org.apache.skywalking.apm.collector.core.module.ModuleManager; @@ -67,6 +68,7 @@ public class SegmentParse { this.spanListeners = new LinkedList<>(); } + @GraphComputingMetric(name = "/segment/parse") public boolean parse(UpstreamSegment segment, ISegmentParseService.Source source) { createSpanListeners(); @@ -95,6 +97,7 @@ public class SegmentParse { return false; } + @GraphComputingMetric(name = "/segment/parse/preBuild") private boolean preBuild(List traceIds, SegmentDecorator segmentDecorator) { StringBuilder segmentIdBuilder = new StringBuilder(); @@ -161,6 +164,7 @@ public class SegmentParse { return true; } + @GraphComputingMetric(name = "/segment/parse/buildSegment") private void buildSegment(String id, byte[] dataBinary) { Segment segment = new Segment(); segment.setId(id); @@ -170,6 +174,7 @@ public class SegmentParse { graph.start(segment); } + @GraphComputingMetric(name = "/segment/parse/bufferFile/write") private void writeToBufferFile(String id, UpstreamSegment upstreamSegment) { logger.debug("push to segment buffer write worker, id: {}", id); SegmentStandardization standardization = new SegmentStandardization(id); diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java index da2e8107b..8d5e43cd3 100644 --- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java +++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java @@ -20,6 +20,8 @@ package org.apache.skywalking.apm.collector.analysis.worker.model.base; import java.util.Iterator; import java.util.List; +import org.apache.skywalking.apm.collector.core.annotations.trace.BatchParameter; +import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric; import org.apache.skywalking.apm.collector.core.data.QueueData; import org.apache.skywalking.apm.collector.core.graph.NodeProcessor; import org.apache.skywalking.apm.collector.core.queue.EndOfBatchContext; @@ -45,7 +47,9 @@ public class LocalAsyncWorkerRef data) { + @GraphComputingMetric(name = "/worker/async/consume") + @Override + public void consume(@BatchParameter List data) { Iterator inputIterator = data.iterator(); int i = 0; diff --git a/apm-collector/apm-collector-boot/pom.xml b/apm-collector/apm-collector-boot/pom.xml index 74832f2fb..d3a2ac1d5 100644 --- a/apm-collector/apm-collector-boot/pom.xml +++ b/apm-collector/apm-collector-boot/pom.xml @@ -175,6 +175,14 @@ ${project.version} + + + + org.apache.skywalking + apm-collector-instrument + ${project.version} + provided + diff --git a/apm-collector/apm-collector-boot/src/main/assembly/assembly.xml b/apm-collector/apm-collector-boot/src/main/assembly/assembly.xml index ea2ee62ac..954e720d4 100644 --- a/apm-collector/apm-collector-boot/src/main/assembly/assembly.xml +++ b/apm-collector/apm-collector-boot/src/main/assembly/assembly.xml @@ -53,5 +53,12 @@ /config + + ${project.basedir}/../apm-collector-instrument/target + + collector-instrument-agent.jar + + /agent + diff --git a/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/BatchParameter.java b/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/BatchParameter.java new file mode 100644 index 000000000..e49570db6 --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/BatchParameter.java @@ -0,0 +1,32 @@ +/* + * 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.collector.core.annotations.trace; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * @author wusheng + */ +@Target(ElementType.PARAMETER) +@Retention(RetentionPolicy.RUNTIME) +public @interface BatchParameter { +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/GraphComputingMetric.java b/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/GraphComputingMetric.java new file mode 100644 index 000000000..d5dec9a86 --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/annotations/trace/GraphComputingMetric.java @@ -0,0 +1,38 @@ +/* + * 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.collector.core.annotations.trace; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * The method with this annotation should be traced, + * and the metrics(avg response time, call count, success rate) could be collected by the instrument agent. + * + * This is an optional annotation. + * + * @author wusheng + */ +@Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +public @interface GraphComputingMetric { + String name(); +} diff --git a/apm-collector/apm-collector-core/src/test/java/org/apache/skywalking/apm/collector/core/annotations/trace/TracedGraphElement.java b/apm-collector/apm-collector-core/src/test/java/org/apache/skywalking/apm/collector/core/annotations/trace/TracedGraphElement.java new file mode 100644 index 000000000..cf95fa346 --- /dev/null +++ b/apm-collector/apm-collector-core/src/test/java/org/apache/skywalking/apm/collector/core/annotations/trace/TracedGraphElement.java @@ -0,0 +1,39 @@ +/* + * 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.collector.core.annotations.trace; + +import java.util.List; + +/** + * This is an example about how to use the tracing annotation in collector. + * These annotations effect only in `-instrument` mode active. + * + * @author wusheng + */ +public class TracedGraphElement { + @GraphComputingMetric(name = "/traced/element/run") + public void run() { + + } + + @GraphComputingMetric(name = "/traced/element/runWithBatch") + public void runWithBatch(@BatchParameter List data) { + + } +} diff --git a/apm-collector/apm-collector-instrument/pom.xml b/apm-collector/apm-collector-instrument/pom.xml index ffd5bb06b..732c49de7 100644 --- a/apm-collector/apm-collector-instrument/pom.xml +++ b/apm-collector/apm-collector-instrument/pom.xml @@ -32,6 +32,7 @@ org.apache.skywalking apm-collector-core ${project.version} + provided net.bytebuddy @@ -39,4 +40,44 @@ 1.7.8 + + + collector-instrument-agent + + + org.apache.maven.plugins + maven-shade-plugin + 2.4.1 + + + package + + shade + + + false + true + true + true + + + + org.apache.skywalking.apm.collector.instrument.CollectorInstrumentAgent + + + + + + com.*:* + org.slf4j:* + org.apache.logging.log4j:* + + + + + + + + diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/CollectorInstrumentAgent.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/CollectorInstrumentAgent.java new file mode 100644 index 000000000..53d15531d --- /dev/null +++ b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/CollectorInstrumentAgent.java @@ -0,0 +1,90 @@ +/* + * 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.collector.instrument; + +import java.lang.instrument.Instrumentation; +import net.bytebuddy.agent.builder.AgentBuilder; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.dynamic.DynamicType; +import net.bytebuddy.implementation.MethodDelegation; +import net.bytebuddy.matcher.ElementMatcher; +import net.bytebuddy.utility.JavaModule; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static net.bytebuddy.matcher.ElementMatchers.declaresMethod; +import static net.bytebuddy.matcher.ElementMatchers.isAnnotatedWith; +import static net.bytebuddy.matcher.ElementMatchers.named; + +/** + * There are a lot of monitoring requirements in collector side. + * The agent way is easy, pluggable, and match the target services/graph-nodes automatically. + * + * This agent is designed and expected running in the same class loader of the collector application, + * so I will keep all class loader issue out of concern, + * in order to keep the trace and monitor codes as simple as possible. + * + * @author wu-sheng + */ +public class CollectorInstrumentAgent { + private final static Logger logger = LoggerFactory.getLogger(CollectorInstrumentAgent.class); + + public static void premain(String agentArgs, Instrumentation instrumentation) { + new AgentBuilder.Default().type( + declaresMethod(isAnnotationedMatch()) + ).transform((builder, typeDescription, classLoader, module) -> { + builder = builder.method(isAnnotationedMatch()) + .intercept(MethodDelegation.withDefaultConfiguration() + .to(new ServiceMetricTracing())); + return builder; + }).with(new AgentBuilder.Listener() { + @Override + public void onDiscovery(String typeName, ClassLoader classLoader, JavaModule module, boolean loaded) { + + } + + @Override + public void onTransformation(TypeDescription typeDescription, ClassLoader classLoader, JavaModule module, + boolean loaded, DynamicType dynamicType) { + + } + + @Override + public void onIgnored(TypeDescription typeDescription, ClassLoader classLoader, JavaModule module, + boolean loaded) { + + } + + @Override public void onError(String typeName, ClassLoader classLoader, JavaModule module, boolean loaded, + Throwable throwable) { + logger.error("Enhance service " + typeName + " error.", throwable); + } + + @Override + public void onComplete(String typeName, ClassLoader classLoader, JavaModule module, boolean loaded) { + + } + }).installOn(instrumentation); + } + + private static ElementMatcher isAnnotationedMatch() { + return isAnnotatedWith(named("org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric")); + } +} diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricCollector.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricCollector.java deleted file mode 100644 index 971591475..000000000 --- a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricCollector.java +++ /dev/null @@ -1,172 +0,0 @@ -/* - * 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.collector.instrument; - -import java.lang.reflect.Method; -import java.util.HashMap; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * The MetricCollector collects the service metrics by Module/Provider/Service structure. - */ -public enum MetricCollector implements Runnable { - INSTANCE; - - private final Logger logger = LoggerFactory.getLogger(MetricCollector.class); - private HashMap modules = new HashMap<>(); - - MetricCollector() { - ScheduledExecutorService service = Executors - .newSingleThreadScheduledExecutor(); - service.scheduleAtFixedRate(this, 10, 60, TimeUnit.SECONDS); - } - - @Override - public void run() { - if (!logger.isDebugEnabled()) { - return; - } - StringBuilder report = new StringBuilder(); - report.append("\n"); - report.append("##################################################################################################################\n"); - report.append("# Collector Service Report #\n"); - report.append("##################################################################################################################\n"); - modules.forEach((moduleName, moduleMetric) -> { - report.append(moduleName).append(":\n"); - moduleMetric.providers.forEach((providerName, providerMetric) -> { - report.append("\t").append(providerName).append(":\n"); - providerMetric.services.forEach((serviceName, serviceMetric) -> { - serviceMetric.methodMetrics.forEach((method, metric) -> { - report.append("\t\t").append(method).append(":\n"); - report.append("\t\t\t").append(metric).append("\n"); - serviceMetric.methodMetrics.put(method, new ServiceMethodMetric()); - }); - }); - }); - }); - - logger.debug(report.toString()); - - } - - ServiceMetric registerService(String module, String provider, String service) { - return initIfAbsent(module).initIfAbsent(provider).initIfAbsent(service); - } - - private ModuleMetric initIfAbsent(String moduleName) { - if (!modules.containsKey(moduleName)) { - ModuleMetric metric = new ModuleMetric(moduleName); - modules.put(moduleName, metric); - return metric; - } - return modules.get(moduleName); - } - - private class ModuleMetric { - private String moduleName; - private HashMap providers = new HashMap<>(); - - public ModuleMetric(String moduleName) { - this.moduleName = moduleName; - } - - private ProviderMetric initIfAbsent(String providerName) { - if (!providers.containsKey(providerName)) { - ProviderMetric metric = new ProviderMetric(providerName); - providers.put(providerName, metric); - return metric; - } - return providers.get(providerName); - } - } - - private class ProviderMetric { - private String providerName; - private HashMap services = new HashMap<>(); - - public ProviderMetric(String providerName) { - this.providerName = providerName; - } - - private ServiceMetric initIfAbsent(String serviceName) { - if (!services.containsKey(serviceName)) { - ServiceMetric metric = new ServiceMetric(serviceName); - services.put(serviceName, metric); - return metric; - } - return services.get(serviceName); - } - } - - class ServiceMetric { - private String serviceName; - private ConcurrentHashMap methodMetrics = new ConcurrentHashMap<>(); - - public ServiceMetric(String serviceName) { - this.serviceName = serviceName; - } - - void trace(Method method, long nano, boolean occurException) { - if (logger.isDebugEnabled()) { - ServiceMethodMetric metric = methodMetrics.get(method); - if (metric == null) { - ServiceMethodMetric methodMetric = new ServiceMethodMetric(); - methodMetrics.putIfAbsent(method, methodMetric); - metric = methodMetrics.get(method); - } - metric.add(nano, occurException); - } - } - } - - private class ServiceMethodMetric { - private AtomicLong totalTimeNano; - private AtomicLong counter; - private AtomicLong errorCounter; - - public ServiceMethodMetric() { - totalTimeNano = new AtomicLong(0); - counter = new AtomicLong(0); - errorCounter = new AtomicLong(0); - } - - private void add(long nano, boolean occurException) { - totalTimeNano.addAndGet(nano); - counter.incrementAndGet(); - if (occurException) - errorCounter.incrementAndGet(); - } - - @Override public String toString() { - if (counter.longValue() == 0) { - return "Avg=N/A"; - } - return "Avg=" + (totalTimeNano.longValue() / counter.longValue()) + " (nano)" + - ", Success Rate=" + (counter.longValue() - errorCounter.longValue()) * 100 / counter.longValue() + - "%"; - } - } -} diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricTree.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricTree.java new file mode 100644 index 000000000..b0b5a77d8 --- /dev/null +++ b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/MetricTree.java @@ -0,0 +1,160 @@ +/* + * 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.collector.instrument; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.skywalking.apm.collector.core.annotations.trace.BatchParameter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author wusheng + */ +public enum MetricTree implements Runnable { + INSTANCE; + private final Logger logger = LoggerFactory.getLogger(MetricTree.class); + + private ScheduledFuture scheduledFuture; + private List metrics = new LinkedList<>(); + private String lineSeparator = System.getProperty("line.separator"); + + MetricTree() { + ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); + scheduledFuture = service.scheduleAtFixedRate(this, 60, 60, TimeUnit.SECONDS); + } + + synchronized MetricNode lookup(String metricName) { + MetricNode node = new MetricNode(metricName); + metrics.add(node); + return node; + } + + @Override + public void run() { + try { + metrics.forEach((metric) -> { + metric.exchange(); + }); + + try { + Thread.sleep(5 * 1000); + } catch (InterruptedException e) { + + } + + StringBuilder logBuffer = new StringBuilder(); + logBuffer.append(lineSeparator); + logBuffer.append("##################################################################################################################").append(lineSeparator); + logBuffer.append("# Collector Service Report #").append(lineSeparator); + logBuffer.append("##################################################################################################################").append(lineSeparator); + metrics.forEach((metric) -> { + metric.toOutput(new ReportWriter() { + + @Override public void writeMetricName(String name) { + logBuffer.append(name).append("").append(lineSeparator); + } + + @Override public void writeMetric(String metrics) { + logBuffer.append("\t"); + logBuffer.append(metrics).append("").append(lineSeparator); + } + }); + }); + + logger.warn(logBuffer.toString()); + } catch (Throwable e) { + logger.error(e.getMessage(), e); + } + } + + class MetricNode { + private String metricName; + private volatile ServiceMetric metric; + + public MetricNode(String metricName) { + this.metricName = metricName; + } + + ServiceMetric getMetric(Method targetMethod, Object[] allArguments) { + if (metric == null) { + synchronized (metricName) { + if (metric == null) { + int detectedBatchIndex = -1; + String batchNodeNameSuffix = null; + if (targetMethod != null) { + Annotation[][] annotations = targetMethod.getParameterAnnotations(); + if (annotations != null) { + int index = 0; + for (Annotation[] parameterAnnotation : annotations) { + if (parameterAnnotation != null) { + for (Annotation annotation : parameterAnnotation) { + if (annotation instanceof BatchParameter) { + detectedBatchIndex = index; + break; + } + } + } + if (detectedBatchIndex > -1) { + break; + } + index++; + } + if (detectedBatchIndex > -1) { + Object listArgs = allArguments[index]; + + if (listArgs instanceof List) { + List args = (List)listArgs; + batchNodeNameSuffix = "/" + args.get(0).getClass().getSimpleName(); + metricName += batchNodeNameSuffix; + } + } + } + } + metric = new ServiceMetric(metricName, detectedBatchIndex); + if (batchNodeNameSuffix != null) { + this.metricName += batchNodeNameSuffix; + } + } + } + } + return metric; + } + + void exchange() { + if (metric != null) { + metric.exchangeWindows(); + } + } + + void toOutput(ReportWriter writer) { + writer.writeMetricName(metricName); + if (metric != null) { + metric.toOutput(writer); + } + + } + } +} diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/TracedService.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ReportWriter.java similarity index 76% rename from apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/TracedService.java rename to apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ReportWriter.java index ffcbda4ab..38f36ebb6 100644 --- a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/TracedService.java +++ b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ReportWriter.java @@ -16,15 +16,14 @@ * */ - package org.apache.skywalking.apm.collector.instrument; /** - * The TracedService implementation are dynamic class, generated by {@link ServiceInstrumentation}. - * - * By that, all the services metrics are collected, and report in the certain cycle through console. - * - * @author wu-sheng + * @author wusheng */ -public interface TracedService { +public interface ReportWriter { + void writeMetricName(String name); + + void writeMetric(String metrics); + } diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceInstrumentation.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceInstrumentation.java deleted file mode 100644 index d73dacd9b..000000000 --- a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceInstrumentation.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * 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.collector.instrument; - -import net.bytebuddy.ByteBuddy; -import net.bytebuddy.description.method.MethodDescription; -import net.bytebuddy.implementation.MethodDelegation; -import net.bytebuddy.matcher.ElementMatcher; -import org.apache.skywalking.apm.collector.core.module.Service; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static net.bytebuddy.matcher.ElementMatchers.isStatic; -import static net.bytebuddy.matcher.ElementMatchers.named; -import static net.bytebuddy.matcher.ElementMatchers.not; - -/** - * The ServiceInstrumentation create the dynamic service implementations based on the provider - * implementation. So the new implementation will report performance metric to {@link MetricCollector}. - * - * @author wu-sheng - */ -public enum ServiceInstrumentation { - INSTANCE; - - private final Logger logger = LoggerFactory.getLogger(ServiceInstrumentation.class); - private ElementMatcher excludeObjectMethodsMatcher; - - public Service buildServiceUnderMonitor(String moduleName, String providerName, Service implementation) { - if (implementation instanceof TracedService) { - // Duplicate service instrument, ignore. - return implementation; - } - try { - return new ByteBuddy().subclass(implementation.getClass()) - .implement(TracedService.class) - .method(getDefaultMatcher()).intercept( - MethodDelegation.withDefaultConfiguration().to(new ServiceMetricTracing(moduleName, providerName, implementation.getClass().getName())) - ).make().load(getClass().getClassLoader() - ).getLoaded().newInstance(); - } catch (InstantiationException e) { - logger.error("Create instrumented service " + implementation.getClass() + " fail.", e); - } catch (IllegalAccessException e) { - logger.error("Create instrumented service " + implementation.getClass() + " fail.", e); - } - return implementation; - } - - private ElementMatcher getDefaultMatcher() { - if (excludeObjectMethodsMatcher == null) { - excludeObjectMethodsMatcher = not(isStatic().or(named("getClass")).or(named("hashCode")).or(named("equals")).or(named("clone")) - .or(named("toString")).or(named("notify")).or(named("notifyAll")).or(named("wait")).or(named("finalize"))); - } - return excludeObjectMethodsMatcher; - } -} diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetric.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetric.java new file mode 100644 index 000000000..afd970aaa --- /dev/null +++ b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetric.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.collector.instrument; + +import java.util.List; + +/** + * @author wusheng + */ +public class ServiceMetric { + private String metricName; + private ServiceMetricRecord winA; + private ServiceMetricRecord winB; + private volatile boolean isUsingWinA; + private volatile int detectedBatchIndex; + + ServiceMetric(String metricName, int detectedBatchIndex) { + this.metricName = metricName; + winA = detectedBatchIndex > -1 ? new ServiceMetricBatchRecord() : new ServiceMetricRecord(); + winB = detectedBatchIndex > -1 ? new ServiceMetricBatchRecord() : new ServiceMetricRecord(); + isUsingWinA = true; + this.detectedBatchIndex = detectedBatchIndex; + } + + public void trace(long nano, boolean occurException, Object[] allArguments) { + ServiceMetricRecord usingRecord = isUsingWinA ? winA : winB; + if (detectedBatchIndex > -1) { + List listArgs = (List)allArguments[detectedBatchIndex]; + ((ServiceMetricBatchRecord)usingRecord).add(nano, occurException, listArgs == null ? 0 : listArgs.size()); + } else { + usingRecord.add(nano, occurException); + } + } + + void exchangeWindows() { + isUsingWinA = !isUsingWinA; + } + + public void toOutput(ReportWriter writer) { + /** + * If using A, then B is available and free to output. + */ + writer.writeMetric(isUsingWinA ? winB.toString() : winA.toString()); + } +} diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricBatchRecord.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricBatchRecord.java new file mode 100644 index 000000000..a9a0cbd5a --- /dev/null +++ b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricBatchRecord.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.collector.instrument; + +import java.util.concurrent.atomic.AtomicLong; + +/** + * @author wusheng + */ +public class ServiceMetricBatchRecord extends ServiceMetricRecord { + private AtomicLong batchRowSize; + + public ServiceMetricBatchRecord() { + super(); + batchRowSize = new AtomicLong(0); + } + + void add(long nano, boolean occurException, int rowSize) { + super.add(nano, occurException); + batchRowSize.addAndGet(rowSize); + } + + @Override void clear() { + super.clear(); + batchRowSize.set(0); + } + + @Override + public String toString() { + if (counter.longValue() == 0) { + return "Avg=N/A"; + } + return super.toString() + " Rows per call = " + (batchRowSize.get() / counter.get()); + } +} diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricRecord.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricRecord.java new file mode 100644 index 000000000..31ade89e2 --- /dev/null +++ b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricRecord.java @@ -0,0 +1,58 @@ +/* + * 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.collector.instrument; + +import java.util.concurrent.atomic.AtomicLong; + +/** + * @author wusheng + */ +public class ServiceMetricRecord { + private AtomicLong totalTimeNano; + protected AtomicLong counter; + private AtomicLong errorCounter; + + public ServiceMetricRecord() { + totalTimeNano = new AtomicLong(0); + counter = new AtomicLong(0); + errorCounter = new AtomicLong(0); + } + + void add(long nano, boolean occurException) { + totalTimeNano.addAndGet(nano); + counter.incrementAndGet(); + if (occurException) + errorCounter.incrementAndGet(); + } + + void clear() { + totalTimeNano.set(0); + counter.set(0); + errorCounter.set(0); + } + + @Override public String toString() { + if (counter.longValue() == 0) { + return "Avg=N/A"; + } + return "Avg=" + (totalTimeNano.longValue() / counter.longValue()) + " (nano)" + + ", Success Rate=" + (counter.longValue() - errorCounter.longValue()) * 100 / counter.longValue() + + "%, Calls=" + counter.longValue(); + } +} diff --git a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricTracing.java b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricTracing.java index 715ec8e46..94629613a 100644 --- a/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricTracing.java +++ b/apm-collector/apm-collector-instrument/src/main/java/org/apache/skywalking/apm/collector/instrument/ServiceMetricTracing.java @@ -16,33 +16,45 @@ * */ - package org.apache.skywalking.apm.collector.instrument; import java.lang.reflect.Method; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; import net.bytebuddy.implementation.bind.annotation.AllArguments; import net.bytebuddy.implementation.bind.annotation.Origin; import net.bytebuddy.implementation.bind.annotation.RuntimeType; import net.bytebuddy.implementation.bind.annotation.SuperCall; import net.bytebuddy.implementation.bind.annotation.This; +import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric; /** * @author wu-sheng */ public class ServiceMetricTracing { - private MetricCollector.ServiceMetric serviceMetric; + private volatile ConcurrentHashMap metrics = new ConcurrentHashMap<>(); - public ServiceMetricTracing(String module, String provider, String service) { - serviceMetric = MetricCollector.INSTANCE.registerService(module, provider, service); + public ServiceMetricTracing() { } @RuntimeType - public Object intercept(@This Object obj, - @AllArguments Object[] allArguments, + public Object intercept( + @This Object inst, @SuperCall Callable zuper, + @AllArguments Object[] allArguments, @Origin Method method ) throws Throwable { + ServiceMetric metric = this.metrics.get(method); + if (metric == null) { + GraphComputingMetric annotation = method.getAnnotation(GraphComputingMetric.class); + String metricName = annotation.name(); + synchronized (inst) { + MetricTree.MetricNode metricNode = MetricTree.INSTANCE.lookup(metricName); + ServiceMetric serviceMetric = metricNode.getMetric(method, allArguments); + metrics.put(method, serviceMetric); + metric = serviceMetric; + } + } boolean occurError = false; long startNano = System.nanoTime(); long endNano; @@ -53,7 +65,8 @@ public class ServiceMetricTracing { throw t; } finally { endNano = System.nanoTime(); - serviceMetric.trace(method, endNano - startNano, occurError); + + metric.trace(endNano - startNano, occurError, allArguments); } } }