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

Adjust log of SegmentServiceClient, close #396

This commit is contained in:
wusheng 2017-09-06 21:38:43 +08:00
parent a46c01e34d
commit 815a47769c
2 changed files with 27 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -21,7 +21,7 @@ public class GRPCStreamServiceStatus {
/**
* @param maxTimeout max wait time, milliseconds.
*/
public void wait4Finish(long maxTimeout) {
public boolean wait4Finish(long maxTimeout) {
long time = 0;
while (!status) {
if (time > maxTimeout) {
@ -30,6 +30,7 @@ public class GRPCStreamServiceStatus {
try2Sleep(5);
time += 5;
}
return status;
}
/**

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

@ -28,6 +28,9 @@ public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSe
private static final ILog logger = LogManager.getLogger(TraceSegmentServiceClient.class);
private static final int TIMEOUT = 30 * 1000;
private long lastLogTime;
private long segmentUplinkedCounter;
private long segmentAbandonedCounter;
private volatile DataCarrier<TraceSegment> carrier;
private volatile TraceSegmentServiceGrpc.TraceSegmentServiceStub serviceStub;
private volatile GRPCChannelStatus status = GRPCChannelStatus.DISCONNECT;
@ -39,6 +42,9 @@ public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSe
@Override
public void boot() throws Throwable {
lastLogTime = System.currentTimeMillis();
segmentUplinkedCounter = 0;
segmentAbandonedCounter = 0;
carrier = new DataCarrier<TraceSegment>(CHANNEL_SIZE, BUFFER_SIZE);
carrier.setBufferStrategy(BufferStrategy.IF_POSSIBLE);
carrier.consume(this, 1);
@ -94,14 +100,27 @@ public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSe
}
upstreamSegmentStreamObserver.onCompleted();
status.wait4Finish(TIMEOUT);
if (logger.isDebugEnable()) {
logger.debug("{} trace segments have been sent to collector.", data.size());
if (status.wait4Finish(TIMEOUT)) {
segmentUplinkedCounter += data.size();
}
} else {
if (logger.isDebugEnable()) {
logger.debug("{} trace segments have been abandoned, cause by no available channel.", data.size());
segmentAbandonedCounter += data.size();
}
printUplinkStatus();
}
private void printUplinkStatus() {
long currentTimeMillis = System.currentTimeMillis();
if (lastLogTime - currentTimeMillis > 30 * 1000) {
lastLogTime = currentTimeMillis;
if (segmentUplinkedCounter > 0) {
logger.debug("{} trace segments have been sent to collector.", segmentUplinkedCounter);
segmentUplinkedCounter = 0;
}
if (segmentAbandonedCounter > 0) {
logger.debug("{} trace segments have been abandoned, cause by no available channel.", segmentAbandonedCounter);
segmentAbandonedCounter = 0;
}
}
}