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

Finish the skeleton codes of sending segments to collector.

This commit is contained in:
wusheng 2017-07-01 21:12:48 +08:00
parent 5621c3b256
commit 95dcd3927b
12 changed files with 227 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,12 +3,14 @@ package org.skywalking.apm.agent.core.boot;
/**
* The <code>BootService</code> is an interface to all remote, which need to boot when plugin mechanism begins to
* work.
* {@link #bootUp()} will be called when <code>BootService</code> start up.
* {@link #boot()} will be called when <code>BootService</code> start up.
*
* @author wusheng
*/
public interface BootService {
void bootUp() throws Throwable;
void beforeBoot() throws Throwable;
void boot() throws Throwable;
void afterBoot() throws Throwable;
}

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

@ -22,6 +22,8 @@ public enum ServiceManager {
public void boot() {
bootedServices = loadAllServices();
beforeBoot();
startup();
afterBoot();
}
@ -36,10 +38,20 @@ public enum ServiceManager {
return bootedServices;
}
private void beforeBoot() {
for (BootService service : bootedServices.values()) {
try {
service.beforeBoot();
} catch (Throwable e) {
logger.error(e, "ServiceManager try to pre-start [{}] fail.", service.getClass().getName());
}
}
}
private void startup() {
for (BootService service : bootedServices.values()) {
try {
service.bootUp();
service.boot();
} catch (Throwable e) {
logger.error(e, "ServiceManager try to start [{}] fail.", service.getClass().getName());
}

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

@ -109,7 +109,12 @@ public class ContextManager implements TracingContextListener, BootService, Igno
}
@Override
public void bootUp() {
public void beforeBoot() throws Throwable {
}
@Override
public void boot() {
TracingContext.ListenerManager.add(this);
IgnoredTracerContext.ListenerManager.add(this);
}

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

@ -1,56 +0,0 @@
package org.skywalking.apm.agent.core.datacarrier;
import java.util.List;
import org.skywalking.apm.agent.core.boot.BootService;
import org.skywalking.apm.agent.core.context.TracingContext;
import org.skywalking.apm.agent.core.context.TracingContextListener;
import org.skywalking.apm.agent.core.context.trace.TraceSegment;
import org.skywalking.apm.agent.core.datacarrier.buffer.BufferStrategy;
import org.skywalking.apm.agent.core.datacarrier.consumer.IConsumer;
import static org.skywalking.apm.agent.core.conf.Config.Buffer.BUFFER_SIZE;
import static org.skywalking.apm.agent.core.conf.Config.Buffer.CHANNEL_SIZE;
/**
* @author wusheng
*/
public class DataBufferService implements BootService, IConsumer<TraceSegment>, TracingContextListener {
private volatile DataCarrier<TraceSegment> carrier;
@Override
public void bootUp() throws Throwable {
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);
}
@Override
public void init() {
}
@Override
public void consume(List<TraceSegment> data) {
}
@Override
public void onError(List<TraceSegment> data, Throwable t) {
}
@Override
public void onExit() {
}
@Override
public void afterFinished(TraceSegment traceSegment) {
carrier.produce(traceSegment);
}
}

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

@ -1,5 +1,6 @@
package org.skywalking.apm.agent.core.discovery;
import org.skywalking.apm.agent.core.boot.BootService;
import org.skywalking.apm.agent.core.remote.DiscoveryRestServiceClient;
/**
@ -7,10 +8,21 @@ import org.skywalking.apm.agent.core.remote.DiscoveryRestServiceClient;
*
* @author wusheng
*/
public class CollectorDiscoveryService extends StatusBootService {
public class CollectorDiscoveryService implements BootService {
@Override
protected void bootUpWithStatus() throws Exception {
public void beforeBoot() throws Throwable {
}
@Override
public void boot() throws Throwable {
Thread collectorClientThread = new Thread(new DiscoveryRestServiceClient(), "collectorClientThread");
collectorClientThread.setDaemon(true);
collectorClientThread.start();
}
@Override
public void afterBoot() throws Throwable {
}
}

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

@ -4,4 +4,5 @@ package org.skywalking.apm.agent.core.remote;
* @author wusheng
*/
public interface GRPCChannelListener {
void statusChanged(GRPCChannelStatus status);
}

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

@ -4,6 +4,9 @@ import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.internal.DnsNameResolverProvider;
import io.grpc.netty.NettyChannelBuilder;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import org.skywalking.apm.agent.core.boot.BootService;
import org.skywalking.apm.agent.core.conf.RemoteDownstreamConfig;
@ -19,9 +22,15 @@ public class GRPCChannelManager implements BootService, Runnable {
private volatile Thread channelManagerThread = null;
private volatile ManagedChannel managedChannel = null;
private Random random = new Random();
private List<GRPCChannelListener> listeners = Collections.synchronizedList(new LinkedList<GRPCChannelListener>());
@Override
public void bootUp() throws Throwable {
public void beforeBoot() throws Throwable {
}
@Override
public void boot() throws Throwable {
this.startupInBackground();
}
@ -61,6 +70,9 @@ public class GRPCChannelManager implements BootService, Runnable {
.maxInboundMessageSize(1024 * 1024 * 50)
.usePlaintext(true);
managedChannel = channelBuilder.build();
for (GRPCChannelListener listener : listeners) {
listener.statusChanged(GRPCChannelStatus.CONNECTED);
}
break;
} catch (Throwable t) {
logger.error(t, "Create channel to {} fail.", server);
@ -73,6 +85,14 @@ public class GRPCChannelManager implements BootService, Runnable {
}
}
public void addChannelListener(GRPCChannelListener listener){
listeners.add(listener);
}
public ManagedChannel getManagedChannel() {
return managedChannel;
}
/**
* Try to sleep, and ignore the {@link InterruptedException}
*

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

@ -0,0 +1,8 @@
package org.skywalking.apm.agent.core.remote;
/**
* @author wusheng
*/
public enum GRPCChannelStatus {
CONNECTED;
}

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

@ -0,0 +1,48 @@
package org.skywalking.apm.agent.core.remote;
/**
* @author wusheng
*/
public class GRPCStreamServiceStatus {
private volatile boolean status;
public GRPCStreamServiceStatus(boolean status) {
this.status = status;
}
public boolean isStatus() {
return status;
}
public void setStatus(boolean status) {
this.status = status;
}
/**
*
* @param maxTimeout max wait time, milliseconds.
*/
public void wait4Finish(long maxTimeout) {
long time = 0;
while (status == false) {
if (time > maxTimeout) {
break;
}
try2Sleep(50);
time += 50;
}
}
/**
* 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) {
}
}
}

107
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
View File

@ -1,14 +1,115 @@
package org.skywalking.apm.agent.core.remote;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
import java.util.List;
import org.skywalking.apm.agent.core.boot.BootService;
import org.skywalking.apm.agent.core.boot.ServiceManager;
import org.skywalking.apm.agent.core.context.TracingContext;
import org.skywalking.apm.agent.core.context.TracingContextListener;
import org.skywalking.apm.agent.core.context.trace.TraceSegment;
import org.skywalking.apm.agent.core.datacarrier.DataCarrier;
import org.skywalking.apm.agent.core.datacarrier.buffer.BufferStrategy;
import org.skywalking.apm.agent.core.datacarrier.consumer.IConsumer;
import org.skywalking.apm.logging.ILog;
import org.skywalking.apm.logging.LogManager;
import org.skywalking.apm.network.collecor.proto.Downstream;
import org.skywalking.apm.network.trace.proto.TraceSegmentServiceGrpc;
import org.skywalking.apm.network.trace.proto.UpstreamSegment;
import static org.skywalking.apm.agent.core.conf.Config.Buffer.BUFFER_SIZE;
import static org.skywalking.apm.agent.core.conf.Config.Buffer.CHANNEL_SIZE;
import static org.skywalking.apm.agent.core.remote.GRPCChannelStatus.CONNECTED;
/**
* @author wusheng
*/
public class TraceSegmentServiceClient {
public void start() {
public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSegment>, TracingContextListener, GRPCChannelListener {
private static final ILog logger = LogManager.getLogger(TraceSegmentServiceClient.class);
private volatile DataCarrier<TraceSegment> carrier;
private volatile TraceSegmentServiceGrpc.TraceSegmentServiceStub serviceStub;
@Override
public void beforeBoot() throws Throwable {
ServiceManager.INSTANCE.findService(GRPCChannelManager.class).addChannelListener(this);
}
@Override
public void boot() throws Throwable {
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);
}
@Override
public void init() {
}
public void switchChannel(){
@Override
public void consume(List<TraceSegment> data) {
final GRPCStreamServiceStatus status = new GRPCStreamServiceStatus(false);
StreamObserver<UpstreamSegment> upstreamSegmentStreamObserver = serviceStub.collect(new StreamObserver<Downstream>() {
@Override
public void onNext(Downstream downstream) {
}
@Override
public void onError(Throwable throwable) {
status.setStatus(true);
}
@Override
public void onCompleted() {
status.setStatus(true);
}
});
try {
for (TraceSegment segment : data) {
//TODO
// segment to PROTOBUF object
upstreamSegmentStreamObserver.onNext(null);
}
} catch (Throwable t) {
logger.error(t, "Send UpstreamSegment to collector fail.");
}
upstreamSegmentStreamObserver.onCompleted();
status.wait4Finish(30 * 1000);
if (logger.isDebugEnable()) {
logger.debug("{} trace segments have been sent to collector.", data.size());
}
}
@Override
public void onError(List<TraceSegment> data, Throwable t) {
logger.error(t, "Try to send {} trace segments to collector, with unexpected exception.", data.size());
}
@Override
public void onExit() {
}
@Override
public void afterFinished(TraceSegment traceSegment) {
carrier.produce(traceSegment);
}
@Override
public void statusChanged(GRPCChannelStatus status) {
if (CONNECTED.equals(status)) {
ManagedChannel channel = ServiceManager.INSTANCE.findService(GRPCChannelManager.class).getManagedChannel();
serviceStub = TraceSegmentServiceGrpc.newStub(channel);
}
}
}

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

@ -28,10 +28,15 @@ public class SamplingService implements BootService {
private volatile ScheduledFuture<?> scheduledFuture;
@Override
public void bootUp() throws Throwable {
public void beforeBoot() throws Throwable {
}
@Override
public void boot() throws Throwable {
if (scheduledFuture != null) {
/**
* If {@link #bootUp()} invokes twice, mostly in test cases,
* If {@link #boot()} invokes twice, mostly in test cases,
* cancel the old one.
*/
scheduledFuture.cancel(true);

2
apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.skywalking.apm.agent.core.boot.BootService
View File

@ -1,4 +1,4 @@
org.skywalking.apm.agent.core.datacarrier.DataBufferService
org.skywalking.apm.agent.core.remote.TraceSegmentServiceClient
org.skywalking.apm.agent.core.context.ContextManager
org.skywalking.apm.agent.core.discovery.CollectorDiscoveryService
org.skywalking.apm.agent.core.sampling.SamplingService