From d960b80fcd1548ddd316bb82f08ce815d2d577f0 Mon Sep 17 00:00:00 2001 From: wusheng Date: Sat, 1 Jul 2017 08:59:41 +0800 Subject: [PATCH] Finish codes about selecting grpc service address and retry. --- .../apm/agent/core/boot/BootService.java | 2 + .../apm/agent/core/boot/ServiceManager.java | 13 ++++- .../agent/core/boot/StatusBootService.java | 32 ------------ .../apm/agent/core/conf/Config.java | 4 +- .../agent/core/context/ContextManager.java | 5 ++ .../core/datacarrier/DataBufferService.java | 4 ++ .../discovery/CollectorDiscoveryService.java | 1 - .../core/remote/GRPCChannelListener.java | 7 +++ .../agent/core/remote/GRPCChannelManager.java | 49 +++++++++++++++---- .../agent/core/sampling/SamplingService.java | 15 ++++-- 10 files changed, 82 insertions(+), 50 deletions(-) delete mode 100644 apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/StatusBootService.java create mode 100644 apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelListener.java diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java index 0bd73f16f..524bff9c7 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java @@ -9,4 +9,6 @@ package org.skywalking.apm.agent.core.boot; */ public interface BootService { void bootUp() throws Throwable; + + void afterBoot() throws Throwable; } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java index ca7f52abd..35a9e1eac 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java @@ -17,12 +17,13 @@ import java.util.ServiceLoader; public enum ServiceManager { INSTANCE; - private static final ILog logger = LogManager.getLogger(StatusBootService.class); + private static final ILog logger = LogManager.getLogger(ServiceManager.class); private Map bootedServices = new HashMap(); public void boot() { bootedServices = loadAllServices(); startup(); + afterBoot(); } private Map loadAllServices() { @@ -45,6 +46,16 @@ public enum ServiceManager { } } + private void afterBoot(){ + for (BootService service : bootedServices.values()) { + try { + service.afterBoot(); + } catch (Throwable e) { + logger.error(e, "Service [{}] AfterBoot process fails.", service.getClass().getName()); + } + } + } + /** * Find a {@link BootService} implementation, which is already started. * diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/StatusBootService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/StatusBootService.java deleted file mode 100644 index 80476c312..000000000 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/StatusBootService.java +++ /dev/null @@ -1,32 +0,0 @@ -package org.skywalking.apm.agent.core.boot; - -/** - * The StatusBootService is an abstract implementations of {@link BootService}, it extends {@link - * BootService}'s ability like this: - *

- * If an implementation extends StatusBootService, it can know whether this service starts up successfully - * or not. It's based on {@link #bootUp()} function finished with or without an exception. if no exception, means start - * up successfully. - * - * @author wusheng - */ -public abstract class StatusBootService implements BootService { - private volatile boolean started = false; - - protected boolean isStarted() { - return this.started; - } - - @Override - public final void bootUp() throws Throwable { - try { - bootUpWithStatus(); - started = true; - } catch (Throwable e) { - started = false; - throw e; - } - } - - protected abstract void bootUpWithStatus() throws Exception; -} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java index 0f5ca4c37..df95a68cc 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java @@ -22,9 +22,9 @@ public class Config { /** * Negative or zero means off, by default. - * {@link #SAMPLE_N_PER_10_SECS} means sampling N {@link TraceSegment} in 10 seconds tops. + * {@link #SAMPLE_N_PER_3_SECS} means sampling N {@link TraceSegment} in 10 seconds tops. */ - public static int SAMPLE_N_PER_10_SECS = -1; + public static int SAMPLE_N_PER_3_SECS = -1; /** * If the operation name of the first span is included in this set, diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java index 188038a0c..9ef71920a 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java @@ -114,6 +114,11 @@ public class ContextManager implements TracingContextListener, BootService, Igno IgnoredTracerContext.ListenerManager.add(this); } + @Override + public void afterBoot() throws Throwable { + + } + @Override public void afterFinished(TraceSegment traceSegment) { CONTEXT.remove(); diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/datacarrier/DataBufferService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/datacarrier/DataBufferService.java index d987e381f..c8c081fb2 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/datacarrier/DataBufferService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/datacarrier/DataBufferService.java @@ -22,6 +22,10 @@ public class DataBufferService implements BootService, IConsumer, carrier = new DataCarrier(CHANNEL_SIZE, BUFFER_SIZE); carrier.setBufferStrategy(BufferStrategy.IF_POSSIBLE); carrier.consume(this, 1); + } + + @Override + public void afterBoot() throws Throwable { TracingContext.ListenerManager.add(this); } diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/discovery/CollectorDiscoveryService.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/discovery/CollectorDiscoveryService.java index eb2d5f4b1..c94885610 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/discovery/CollectorDiscoveryService.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/discovery/CollectorDiscoveryService.java @@ -1,6 +1,5 @@ package org.skywalking.apm.agent.core.discovery; -import org.skywalking.apm.agent.core.boot.StatusBootService; import org.skywalking.apm.agent.core.remote.DiscoveryRestServiceClient; /** diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelListener.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelListener.java new file mode 100644 index 000000000..792d62571 --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelListener.java @@ -0,0 +1,7 @@ +package org.skywalking.apm.agent.core.remote; + +/** + * @author wusheng + */ +public interface GRPCChannelListener { +} 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 f92a91d0f..74570ea14 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 @@ -4,7 +4,9 @@ import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.internal.DnsNameResolverProvider; import io.grpc.netty.NettyChannelBuilder; +import java.util.Random; import org.skywalking.apm.agent.core.boot.BootService; +import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig; import org.skywalking.apm.logging.ILog; import org.skywalking.apm.logging.LogManager; @@ -16,12 +18,18 @@ public class GRPCChannelManager implements BootService, Runnable { private volatile Thread channelManagerThread = null; private volatile ManagedChannel managedChannel = null; + private Random random = new Random(); @Override public void bootUp() throws Throwable { this.startupInBackground(); } + @Override + public void afterBoot() throws Throwable { + + } + private void startupInBackground() { if (channelManagerThread == null || !channelManagerThread.isAlive()) { synchronized (this) { @@ -41,17 +49,40 @@ public class GRPCChannelManager implements BootService, Runnable { @Override public void run() { - ManagedChannelBuilder channelBuilder = - NettyChannelBuilder.forAddress("127.0.0.1", 808) - .nameResolverFactory(new DnsNameResolverProvider()) - .maxInboundMessageSize(1024 * 1024 * 50) - .usePlaintext(true); - managedChannel = channelBuilder.build(); + while (true) { + if (RemoteDownstreamConfig.Collector.GRPC_SERVERS.size() > 0) { + int index = random.nextInt() % RemoteDownstreamConfig.Collector.GRPC_SERVERS.size() + String server = RemoteDownstreamConfig.Collector.GRPC_SERVERS.get(index); + try { + String[] ipAndPort = server.split(":"); + ManagedChannelBuilder channelBuilder = + NettyChannelBuilder.forAddress(ipAndPort[0], Integer.parseInt(ipAndPort[1])) + .nameResolverFactory(new DnsNameResolverProvider()) + .maxInboundMessageSize(1024 * 1024 * 50) + .usePlaintext(true); + managedChannel = channelBuilder.build(); + break; + } catch (Throwable t) { + logger.error(t, "Create channel to {} fail.", server); + } + } + + int waitTime = 5 * 1000; + logger.debug("Selected collector grpc service is not available. Wait {} seconds to try", waitTime); + try2Sleep(waitTime); + } } - public static void main(String[] args) throws Throwable { - new GRPCChannelManager().bootUp(); + /** + * Try to sleep, and ignore the {@link InterruptedException} + * + * @param millis the length of time to sleep in milliseconds + */ + private void try2Sleep(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { - Thread.sleep(60 * 1000); + } } } 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 6d0b745c1..993a3e4ce 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 @@ -16,7 +16,7 @@ import org.skywalking.apm.logging.LogManager; * have been traced, but, considering CPU cost of serialization/deserialization, and network bandwidth, the agent do NOT * send all of them to collector, if SAMPLING is on. *

- * By default, SAMPLING is on, and {@see {@link Config.Agent#SAMPLE_N_PER_10_SECS}} + * By default, SAMPLING is on, and {@see {@link Config.Agent#SAMPLE_N_PER_3_SECS }} * * @author wusheng */ @@ -36,7 +36,7 @@ public class SamplingService implements BootService { */ scheduledFuture.cancel(true); } - if (Config.Agent.SAMPLE_N_PER_10_SECS > 0) { + if (Config.Agent.SAMPLE_N_PER_3_SECS > 0) { on = true; this.resetSamplingFactor(); ScheduledExecutorService service = Executors @@ -46,18 +46,23 @@ public class SamplingService implements BootService { public void run() { resetSamplingFactor(); } - }, 0, 10, TimeUnit.SECONDS); - logger.debug("Agent sampling mechanism started. Sample {} traces in 10 seconds.", Config.Agent.SAMPLE_N_PER_10_SECS); + }, 0, 3, TimeUnit.SECONDS); + logger.debug("Agent sampling mechanism started. Sample {} traces in 10 seconds.", Config.Agent.SAMPLE_N_PER_3_SECS); } } + @Override + public void afterBoot() throws Throwable { + + } + /** * @return true, if sampling mechanism is on, and get the sampling factor successfully. */ public boolean trySampling() { if (on) { int factor = samplingFactorHolder.get(); - if (factor < Config.Agent.SAMPLE_N_PER_10_SECS) { + if (factor < Config.Agent.SAMPLE_N_PER_3_SECS) { boolean success = samplingFactorHolder.compareAndSet(factor, factor + 1); return success; } else {