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 524bff9c7..d3538f915 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 @@ -3,12 +3,14 @@ package org.skywalking.apm.agent.core.boot; /** * The BootService is an interface to all remote, which need to boot when plugin mechanism begins to * work. - * {@link #bootUp()} will be called when BootService start up. + * {@link #boot()} will be called when BootService start up. * * @author wusheng */ public interface BootService { - void bootUp() throws Throwable; + void beforeBoot() throws Throwable; + + void boot() 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 35a9e1eac..06283a8ef 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 @@ -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()); } 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 9ef71920a..13a28e9d8 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 @@ -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); } 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 deleted file mode 100644 index c8c081fb2..000000000 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/datacarrier/DataBufferService.java +++ /dev/null @@ -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, TracingContextListener { - private volatile DataCarrier carrier; - - @Override - public void bootUp() throws Throwable { - 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); - } - - @Override - public void init() { - - } - - @Override - public void consume(List data) { - - } - - @Override - public void onError(List data, Throwable t) { - - } - - @Override - public void onExit() { - - } - - @Override - public void afterFinished(TraceSegment traceSegment) { - carrier.produce(traceSegment); - } -} 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 c94885610..6022a700f 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,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 { + + } } 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 index 792d62571..3b3887395 100644 --- 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 @@ -4,4 +4,5 @@ package org.skywalking.apm.agent.core.remote; * @author wusheng */ public interface GRPCChannelListener { + void statusChanged(GRPCChannelStatus status); } 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 6dd129fcb..93899e4cd 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,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 listeners = Collections.synchronizedList(new LinkedList()); @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} * diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelStatus.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelStatus.java new file mode 100644 index 000000000..78060a69a --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCChannelStatus.java @@ -0,0 +1,8 @@ +package org.skywalking.apm.agent.core.remote; + +/** + * @author wusheng + */ +public enum GRPCChannelStatus { + CONNECTED; +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCStreamServiceStatus.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCStreamServiceStatus.java new file mode 100644 index 000000000..0717e59ac --- /dev/null +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/GRPCStreamServiceStatus.java @@ -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) { + + } + } +} diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java index 76bc6cc58..cbc312782 100644 --- a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java +++ b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java @@ -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, TracingContextListener, GRPCChannelListener { + private static final ILog logger = LogManager.getLogger(TraceSegmentServiceClient.class); + + private volatile DataCarrier 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(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 data) { + final GRPCStreamServiceStatus status = new GRPCStreamServiceStatus(false); + StreamObserver upstreamSegmentStreamObserver = serviceStub.collect(new StreamObserver() { + @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 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); + } } } 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 993a3e4ce..8f3adab55 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 @@ -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); diff --git a/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.skywalking.apm.agent.core.boot.BootService b/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.skywalking.apm.agent.core.boot.BootService index 1b0c29fba..91b545ba1 100644 --- a/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.skywalking.apm.agent.core.boot.BootService +++ b/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.skywalking.apm.agent.core.boot.BootService @@ -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