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

Add agent-client codes.

This commit is contained in:
wusheng 2017-03-17 17:38:03 +08:00
parent ebb0f9e469
commit 6a833baf75
8 changed files with 49 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/Context.java
View File

@ -9,7 +9,7 @@ public interface Context {
void putProvider(AbstractWorkerProvider provider) throws UsedRoleNameException;
WorkerRefs lookup(Role role) throws WorkerNotFountException;
WorkerRefs lookup(Role role) throws WorkerNotFoundException;
void put(WorkerRef workerRef);

4
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkerContext.java
View File

@ -13,12 +13,12 @@ public abstract class WorkerContext implements Context {
private Map<String, List<WorkerRef>> roleWorkers = new ConcurrentHashMap<>();
@Override
final public WorkerRefs lookup(Role role) throws WorkerNotFountException {
final public WorkerRefs lookup(Role role) throws WorkerNotFoundException {
if (roleWorkers.containsKey(role.roleName())) {
WorkerRefs refs = new WorkerRefs(roleWorkers.get(role.roleName()), role.workerSelector());
return refs;
} else {
throw new WorkerNotFountException("role=" + role.roleName() + ", no available worker.");
throw new WorkerNotFoundException("role=" + role.roleName() + ", no available worker.");
}
}

7
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkerNotFoundException.java Normal file
View File

@ -0,0 +1,7 @@
package com.a.eye.skywalking.collector.actor;
public class WorkerNotFoundException extends Exception {
public WorkerNotFoundException(String message){
super(message);
}
}

7
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkerNotFountException.java
View File

@ -1,7 +0,0 @@
package com.a.eye.skywalking.collector.actor;
public class WorkerNotFountException extends Exception {
public WorkerNotFountException(String message){
super(message);
}
}

5
skywalking-sniffer/skywalking-api/pom.xml
View File

@ -35,6 +35,11 @@
<artifactId>skywalking-collector-cluster</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-collector-role</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>net.bytebuddy</groupId>
<artifactId>byte-buddy</artifactId>

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/api/boot/BootService.java
View File

@ -8,5 +8,5 @@ package com.a.eye.skywalking.api.boot;
* @author wusheng
*/
public interface BootService {
void bootUp();
void bootUp() throws Exception;
}

4
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/api/boot/StatusBootService.java
View File

@ -18,7 +18,7 @@ public abstract class StatusBootService implements BootService {
}
@Override
public final void bootUp(){
public final void bootUp() throws Exception{
try {
bootUpWithStatus();
started = true;
@ -28,5 +28,5 @@ public abstract class StatusBootService implements BootService {
}
}
protected abstract void bootUpWithStatus();
protected abstract void bootUpWithStatus() throws Exception;
}

42
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/api/client/CollectorClientService.java
View File

@ -1,8 +1,13 @@
package com.a.eye.skywalking.api.client;
import com.a.eye.skywalking.api.boot.BootService;
import com.a.eye.skywalking.api.boot.ServiceManager;
import com.a.eye.skywalking.api.boot.StatusBootService;
import com.a.eye.skywalking.api.queue.TraceSegmentProcessQueue;
import com.a.eye.skywalking.collector.CollectorSystem;
import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
import com.a.eye.skywalking.collector.actor.WorkerNotFoundException;
import com.a.eye.skywalking.collector.actor.WorkerRefs;
import com.a.eye.skywalking.collector.role.TraceSegmentReceiverRole;
import com.a.eye.skywalking.logging.ILog;
import com.a.eye.skywalking.logging.LogManager;
import com.a.eye.skywalking.trace.TraceSegment;
@ -11,35 +16,50 @@ import java.util.List;
/**
* @author wusheng
*/
public class CollectorClientService implements Runnable, BootService {
public class CollectorClientService extends StatusBootService implements Runnable {
private static ILog logger = LogManager.getLogger(CollectorClientService.class);
private static long NO_DATA_SLEEP_TIME_MILLIS = 500;
private static long SLEEP_TIME_MILLIS = 500;
private ClusterWorkerContext clusterContext;
/**
* Start a new {@link Thread} to get finished {@link TraceSegment} by {@link TraceSegmentProcessQueue#getCachedTraceSegments()}
*/
@Override
public void bootUp() {
protected void bootUpWithStatus() throws Exception {
Thread collectorClientThread = new Thread(this, "collectorClientThread");
collectorClientThread.start();
CollectorSystem collectorSystem = new CollectorSystem();
collectorSystem.boot();
clusterContext = collectorSystem.getClusterContext();
}
@Override
public void run() {
while (true) {
try {
long sleepTime = -1;
TraceSegmentProcessQueue segmentProcessQueue = ServiceManager.INSTANCE.findService(TraceSegmentProcessQueue.class);
List<TraceSegment> cachedTraceSegments = segmentProcessQueue.getCachedTraceSegments();
if (cachedTraceSegments.size() > 0) {
for (TraceSegment segment : cachedTraceSegments) {
//TODO: wusheng
//send data
try {
WorkerRefs workerRefs = clusterContext.lookup(TraceSegmentReceiverRole.INSTANCE);
workerRefs.tell(segment);
}catch (WorkerNotFoundException t){
logger.error(t, "Role={} not found.", TraceSegmentReceiverRole.INSTANCE.roleName());
/**
* No receiver found, means collector server is off-line.
*/
sleepTime = SLEEP_TIME_MILLIS * 10;
break;
}
}
/**
* No sleep, when exist finished {@link TraceSegment}.
*/
} else {
try2Sleep(NO_DATA_SLEEP_TIME_MILLIS);
}else{
sleepTime = SLEEP_TIME_MILLIS;
}
if(sleepTime > 0){
try2Sleep(sleepTime);
}
} catch (Throwable t) {
logger.error(t, "Send trace segments to collector failure.");