diff --git a/README_ZH.md b/README_ZH.md index 053748ddd..27414f1cd 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -12,7 +12,7 @@ Sky Walking | [English](README.md) * 自动java探针，**不需要修改应用程序源代码** - * 高性能探针，针对单实例5000tps的应用，在**不需要采样的情况下**，只增加**10%**的CPU开销。 +  * 高性能探针，针对单实例5000tps的应用，在**全量采集的情况下**，只增加**10%**的CPU开销。 * [中间件，框架与类库支持列表](https://github.com/wu-sheng/sky-walking/wiki/3.2-supported-list). * 手动探针 * [使用OpenTracing手动探针API](http://opentracing.io/documentation/pages/supported-tracers) diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/cost/SegmentCostSpanListener.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/cost/SegmentCostSpanListener.java index 179d23c54..ace252874 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/cost/SegmentCostSpanListener.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/cost/SegmentCostSpanListener.java @@ -39,7 +39,7 @@ import org.slf4j.LoggerFactory; /** * @author pengys5 */ -public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListener, LocalSpanListener, FirstSpanListener { +public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListener, LocalSpanListener,FirstSpanListener { private final Logger logger = LoggerFactory.getLogger(SegmentCostSpanListener.class); @@ -50,10 +50,7 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe @Override public void parseFirst(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { timeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanObject.getStartTime()); - } - @Override - public void parseEntry(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { SegmentCostDataDefine.SegmentCost segmentCost = new SegmentCostDataDefine.SegmentCost(); segmentCost.setSegmentId(segmentId); segmentCost.setApplicationId(applicationId); @@ -71,6 +68,11 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe isError = isError || spanObject.getIsError(); } + @Override + public void parseEntry(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { + isError = isError || spanObject.getIsError(); + } + @Override public void parseExit(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { isError = isError || spanObject.getIsError(); @@ -96,4 +98,4 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe } } } -} \ No newline at end of file +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/DefaultNamedThreadFactory.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/DefaultNamedThreadFactory.java new file mode 100644 index 000000000..55395d810 --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/DefaultNamedThreadFactory.java @@ -0,0 +1,39 @@ +/* + * 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.agent.core.boot; + +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * @author zhangkewei + */ +public class DefaultNamedThreadFactory implements ThreadFactory { + private static final AtomicInteger BOOT_SERVICE_SEQ = new AtomicInteger(0); + private final AtomicInteger threadSeq = new AtomicInteger(0); + private final String namePrefix; + public DefaultNamedThreadFactory(String name) { + namePrefix = "SkywalkingAgent-" + BOOT_SERVICE_SEQ.incrementAndGet() + "-" + name + "-"; + } + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r,namePrefix + threadSeq.getAndIncrement()); + t.setDaemon(true); + return t; + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/NoopSpan.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/NoopSpan.java index 716beb72c..8f129d643 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/NoopSpan.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/NoopSpan.java @@ -39,7 +39,7 @@ public class NoopSpan implements AbstractSpan { } @Override public AbstractSpan errorOccurred() { - return null; + return this; } public void finish() { diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java index 633dfc838..2977e8ea0 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/jvm/JVMService.java @@ -25,6 +25,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.skywalking.apm.agent.core.boot.BootService; +import org.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; import org.skywalking.apm.agent.core.boot.ServiceManager; import org.skywalking.apm.agent.core.conf.Config; import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig; @@ -57,7 +58,6 @@ public class JVMService implements BootService, Runnable { private volatile ScheduledFuture collectMetricFuture; private volatile ScheduledFuture sendMetricFuture; private Sender sender; - @Override public void beforeBoot() throws Throwable { queue = new LinkedBlockingQueue(Config.Jvm.BUFFER_SIZE); @@ -68,10 +68,10 @@ public class JVMService implements BootService, Runnable { @Override public void boot() throws Throwable { collectMetricFuture = Executors - .newSingleThreadScheduledExecutor() + .newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("JVMService-produce")) .scheduleAtFixedRate(this, 0, 1, TimeUnit.SECONDS); sendMetricFuture = Executors - .newSingleThreadScheduledExecutor() + .newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("JVMService-consume")) .scheduleAtFixedRate(sender, 0, 1, TimeUnit.SECONDS); } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java index f0e192384..0149fceaa 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java @@ -24,6 +24,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.skywalking.apm.agent.core.boot.BootService; +import org.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; import org.skywalking.apm.agent.core.boot.ServiceManager; import org.skywalking.apm.agent.core.conf.Config; import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig; @@ -89,7 +90,7 @@ public class AppAndServiceRegisterClient implements BootService, GRPCChannelList @Override public void boot() throws Throwable { applicationRegisterFuture = Executors - .newSingleThreadScheduledExecutor() + .newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("AppAndServiceRegisterClient")) .scheduleAtFixedRate(this, 0, Config.Collector.APP_AND_SERVICE_REGISTER_CHECK_INTERVAL, TimeUnit.SECONDS); } @@ -105,6 +106,7 @@ public class AppAndServiceRegisterClient implements BootService, GRPCChannelList @Override public void run() { + logger.debug("AppAndServiceRegisterClient running, status:{}.",status); boolean shouldTry = true; while (CONNECTED.equals(status) && shouldTry) { shouldTry = false; diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java index b8a042831..434df7404 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/CollectorDiscoveryService.java @@ -22,6 +22,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.skywalking.apm.agent.core.boot.BootService; +import org.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; import org.skywalking.apm.agent.core.conf.Config; /** @@ -39,7 +40,7 @@ public class CollectorDiscoveryService implements BootService { @Override public void boot() throws Throwable { - future = Executors.newSingleThreadScheduledExecutor() + future = Executors.newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("CollectorDiscoveryService")) .scheduleAtFixedRate(new DiscoveryRestServiceClient(), 0, Config.Collector.DISCOVERY_CHECK_INTERVAL, TimeUnit.SECONDS); } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java index 687d2391d..8f5e264ef 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java @@ -32,12 +32,12 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.skywalking.apm.agent.core.boot.BootService; +import org.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; +import org.skywalking.apm.agent.core.conf.Config; import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig; import org.skywalking.apm.logging.ILog; import org.skywalking.apm.logging.LogManager; -import static org.skywalking.apm.agent.core.conf.Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL; - /** * @author wusheng */ @@ -58,8 +58,8 @@ public class GRPCChannelManager implements BootService, Runnable { @Override public void boot() throws Throwable { connectCheckFuture = Executors - .newSingleThreadScheduledExecutor() - .scheduleAtFixedRate(this, 0, GRPC_CHANNEL_CHECK_INTERVAL, TimeUnit.SECONDS); + .newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("GRPCChannelManager")) + .scheduleAtFixedRate(this, 0, Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL, TimeUnit.SECONDS); } @Override @@ -71,15 +71,18 @@ public class GRPCChannelManager implements BootService, Runnable { public void shutdown() throws Throwable { connectCheckFuture.cancel(true); managedChannel.shutdownNow(); + logger.debug("Selected collector grpc service shutdown."); } @Override public void run() { + logger.debug("Selected collector grpc service running, reconnect:{}.",reconnect); if (reconnect) { if (RemoteDownstreamConfig.Collector.GRPC_SERVERS.size() > 0) { - int index = random.nextInt() % RemoteDownstreamConfig.Collector.GRPC_SERVERS.size(); - String server = RemoteDownstreamConfig.Collector.GRPC_SERVERS.get(index); + String server = ""; try { + int index = Math.abs(random.nextInt()) % RemoteDownstreamConfig.Collector.GRPC_SERVERS.size(); + server = RemoteDownstreamConfig.Collector.GRPC_SERVERS.get(index); String[] ipAndPort = server.split(":"); ManagedChannelBuilder channelBuilder = NettyChannelBuilder.forAddress(ipAndPort[0], Integer.parseInt(ipAndPort[1])) @@ -100,7 +103,7 @@ public class GRPCChannelManager implements BootService, Runnable { } } - logger.debug("Selected collector grpc service is not available. Wait {} seconds to retry", GRPC_CHANNEL_CHECK_INTERVAL); + logger.debug("Selected collector grpc service is not available. Wait {} seconds to retry", Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL); } } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/sampling/SamplingService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/sampling/SamplingService.java index a3658ceb8..e62999425 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/sampling/SamplingService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/sampling/SamplingService.java @@ -24,6 +24,7 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.skywalking.apm.agent.core.boot.BootService; +import org.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory; import org.skywalking.apm.agent.core.conf.Config; import org.skywalking.apm.agent.core.context.trace.TraceSegment; import org.skywalking.apm.logging.ILog; @@ -63,7 +64,7 @@ public class SamplingService implements BootService { on = true; this.resetSamplingFactor(); ScheduledExecutorService service = Executors - .newSingleThreadScheduledExecutor(); + .newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("SamplingService")); scheduledFuture = service.scheduleAtFixedRate(new Runnable() { @Override public void run() { diff --git a/apm-sniffer/apm-agent-core/src/test/java/org/skywalking/apm/agent/core/boot/DefaultNamedThreadFactoryTest.java b/apm-sniffer/apm-agent-core/src/test/java/org/skywalking/apm/agent/core/boot/DefaultNamedThreadFactoryTest.java new file mode 100644 index 000000000..8d1662767 --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/test/java/org/skywalking/apm/agent/core/boot/DefaultNamedThreadFactoryTest.java @@ -0,0 +1,41 @@ +/* + * 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.agent.core.boot; + +import org.junit.Test; + +import static org.junit.Assert.assertNotNull; + +/** + * @author zhangkewei + */ +public class DefaultNamedThreadFactoryTest { + + @Test + public void testNamedThread() throws Exception { + Thread newThread = new DefaultNamedThreadFactory("DefaultNamedThreadFactoryTest").newThread(new Runnable() { + @Override + public void run() { + + } + }); + newThread.start(); + assertNotNull(newThread.getName()); + assert(newThread.getName().contains("DefaultNamedThreadFactoryTest")); + } +}