kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Finish codes about selecting grpc service address and retry.

This commit is contained in:
wusheng 2017-07-01 08:59:41 +08:00
parent 77745a7589
commit d960b80fcd
10 changed files with 82 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/BootService.java
View File

@ -9,4 +9,6 @@ package org.skywalking.apm.agent.core.boot;
*/
public interface BootService {
void bootUp() throws Throwable;
void afterBoot() throws Throwable;
}

13
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/ServiceManager.java
View File

@ -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<Class, BootService> bootedServices = new HashMap<Class, BootService>();
public void boot() {
bootedServices = loadAllServices();
startup();
afterBoot();
}
private Map<Class, BootService> 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.
*

32
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/boot/StatusBootService.java
View File

@ -1,32 +0,0 @@
package org.skywalking.apm.agent.core.boot;
/**
* The <code>StatusBootService</code> is an abstract implementations of {@link BootService}, it extends {@link
* BootService}'s ability like this:
* <p>
* If an implementation extends <code>StatusBootService</code>, 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;
}

4
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/conf/Config.java
View File

@ -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,

5
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java
View File

@ -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();

4
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/datacarrier/DataBufferService.java
View File

@ -22,6 +22,10 @@ public class DataBufferService implements BootService, IConsumer<TraceSegment>,
carrier = new DataCarrier<TraceSegment>(CHANNEL_SIZE, BUFFER_SIZE);
carrier.setBufferStrategy(BufferStrategy.IF_POSSIBLE);
carrier.consume(this, 1);
}
@Override
public void afterBoot() throws Throwable {
TracingContext.ListenerManager.add(this);
}

1
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/discovery/CollectorDiscoveryService.java
View File

@ -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;
/**

7
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelListener.java Normal file
View File

@ -0,0 +1,7 @@
package org.skywalking.apm.agent.core.remote;
/**
* @author wusheng
*/
public interface GRPCChannelListener {
}

49
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelManager.java
View File

@ -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);
}
}
}

15
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/sampling/SamplingService.java
View File

@ -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.
* <p>
* 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 {