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

* add Agent2RoutingClient, to provie send ReqeustSpan and AckSpan to routing servers.

This commit is contained in:
wusheng 2016-11-27 13:12:53 +08:00
parent c99679a868
commit ec08ca15e5
23 changed files with 247 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
skywalking-network/src/main/java/com/a/eye/skywalking/network/Client.java
View File

@ -21,4 +21,12 @@ public class Client {
public TraceSearchClient newTraceSearchClient(StorageClientListener listener){
return new TraceSearchClient(channel, listener);
}
public void shutdown(){
channel.shutdownNow();
}
public boolean isShutdown(){
return channel.isShutdown() || channel.isTerminated();
}
}

6
skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/client/SpanStorageClient.java
View File

@ -9,6 +9,8 @@ import io.grpc.ManagedChannel;
import io.grpc.stub.CallStreamObserver;
import io.grpc.stub.StreamObserver;
import java.util.List;
public class SpanStorageClient {
private final SpanStorageServiceGrpc.SpanStorageServiceStub spanStorageStub;
@ -20,7 +22,7 @@ public class SpanStorageClient {
this.listener = listener;
}
public void sendRequestSpan(RequestSpan... requestSpan) {
public void sendRequestSpan(List<RequestSpan> requestSpan) {
StreamObserver<RequestSpan> requestSpanStreamObserver =
spanStorageStub.storageRequestSpan(new StreamObserver<SendResult>() {
@Override
@ -46,7 +48,7 @@ public class SpanStorageClient {
requestSpanStreamObserver.onCompleted();
}
public void sendACKSpan(AckSpan... ackSpan) {
public void sendACKSpan(List<AckSpan> ackSpan) {
StreamObserver<AckSpan> ackSpanStreamObserver =
spanStorageStub.storageACKSpan(new StreamObserver<SendResult>() {
@Override

3
skywalking-sniffer/skywalking-agent/src/main/java/com/a/eye/skywalking/agent/SkyWalkingAgent.java
View File

@ -1,6 +1,7 @@
package com.a.eye.skywalking.agent;
import com.a.eye.skywalking.agent.junction.SkyWalkingEnhanceMatcher;
import com.a.eye.skywalking.client.Agent2RoutingClient;
import com.a.eye.skywalking.conf.Config;
import com.a.eye.skywalking.conf.ConfigInitializer;
import com.a.eye.skywalking.logging.EasyLogResolver;
@ -35,6 +36,8 @@ public class SkyWalkingAgent {
logger = LogManager.getLogger(SkyWalkingAgent.class);
initConfig();
Agent2RoutingClient.INSTANCE.onReady();
final PluginDefineCategory pluginDefineCategory = PluginDefineCategory.category(new PluginBootstrap().loadPlugins());
new AgentBuilder.Default().type(enhanceClassMatcher(pluginDefineCategory).and(not(isInterface()))).transform(new AgentBuilder.Transformer() {

159
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/client/Agent2RoutingClient.java Normal file
View File

@ -0,0 +1,159 @@
package com.a.eye.skywalking.client;
import com.a.eye.skywalking.conf.Config;
import com.a.eye.skywalking.disruptor.ack.SendAckSpanEventHandler;
import com.a.eye.skywalking.disruptor.request.SendRequestSpanEventHandler;
import com.a.eye.skywalking.health.report.HealthCollector;
import com.a.eye.skywalking.health.report.HeathReading;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.network.Client;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.SendResult;
import com.a.eye.skywalking.network.grpc.client.SpanStorageClient;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.locks.LockSupport;
/**
* Created by wusheng on 2016/11/27.
*/
public class Agent2RoutingClient extends Thread {
private static ILog logger = LogManager.getLogger(Agent2RoutingClient.class);
private List<ServerAddr> addrList;
private Client client;
private SpanStorageClient spanStorageClient;
private NetworkListener listener;
private SendRequestSpanEventHandler requestSpanDataSupplier = null;
private SendAckSpanEventHandler ackSpanDataSupplier = null;
private volatile boolean connected = false;
public static Agent2RoutingClient INSTANCE = new Agent2RoutingClient();
public Agent2RoutingClient() {
String[] serverList = Config.SkyWalking.SERVERS.split(",");
addrList = new ArrayList<>(serverList.length);
for (String server : serverList) {
String[] addrSegments = server.split(":");
if (addrSegments.length != 2) {
throw new IllegalArgumentException("server addr should like ip:port, illegal addr:" + server);
}
addrList.add(new ServerAddr(addrSegments[0], addrSegments[2]));
}
listener = new NetworkListener();
}
public void onReady() {
this.connect();
this.start();
}
public void setRequestSpanDataSupplier(SendRequestSpanEventHandler requestSpanDataSupplier) {
this.requestSpanDataSupplier = requestSpanDataSupplier;
}
public void setAckSpanDataSupplier(SendAckSpanEventHandler ackSpanDataSupplier) {
this.ackSpanDataSupplier = ackSpanDataSupplier;
}
private void connect() {
try {
if(client != null && !client.isShutdown()){
client.shutdown();
}
int addrIdx = new Random().nextInt(addrList.size());
ServerAddr addr = addrList.get(addrIdx);
client = new Client(addr.ip, addr.port);
spanStorageClient = client.newSpanStorageClient(listener);
connected = true;
} catch (Exception e) {
HealthCollector.getCurrentHeathReading("Agent2RoutingClient").updateData(HeathReading.ERROR, "connect to routing node failure.");
}
}
@Override
public void run() {
while (true) {
try {
while (connected && !client.isShutdown()) {
List<RequestSpan> requestData = this.requestSpanDataSupplier.getBufferData();
List<AckSpan> ackData = this.ackSpanDataSupplier.getBufferData();
if (requestData.size() > 0 || ackData.size() > 0) {
listener.begin();
spanStorageClient.sendRequestSpan(requestData);
spanStorageClient.sendACKSpan(ackData);
while (!listener.isBatchFinished()) {
LockSupport.parkNanos(1);
}
} else {
try {
Thread.sleep(10 * 1000L);
} catch (InterruptedException e) {
}
}
}
try {
Thread.sleep(30 * 1000L);
} catch (InterruptedException e) {
}
this.connect();
} catch (Throwable e) {
logger.error("unexpected failure.", e);
}
}
}
class NetworkListener implements StorageClientListener {
private volatile boolean batchFinished = false;
void begin() {
batchFinished = false;
}
boolean isBatchFinished() {
return batchFinished;
}
@Override
public void onError(Throwable throwable) {
batchFinished = true;
HealthCollector.getCurrentHeathReading("Agent2RoutingClient").updateData(HeathReading.ERROR, "send data to routing node failure.");
}
@Override
public void onBatchFinished(SendResult sendResult) {
batchFinished = true;
HealthCollector.getCurrentHeathReading("Agent2RoutingClient").updateData(HeathReading.INFO, "batch send data to routing node.");
}
}
class ServerAddr {
String ip;
Integer port;
public ServerAddr(String ip, String port) {
this.ip = ip;
try {
this.port = Integer.parseInt(port);
} catch (NumberFormatException e) {
throw new IllegalArgumentException("server addr should like ip:port, illegal port:" + port);
}
}
}
}

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/conf/ConfigInitializer.java
View File

@ -2,7 +2,7 @@ package com.a.eye.skywalking.conf;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.protocol.util.StringUtil;
import com.a.eye.skywalking.util.StringUtil;
import java.io.File;
import java.io.FileInputStream;

38
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/disruptor/ack/SendAckSpanEventHandler.java
View File

@ -1,10 +1,12 @@
package com.a.eye.skywalking.disruptor.ack;
import com.a.eye.skywalking.client.Agent2RoutingClient;
import com.a.eye.skywalking.health.report.HealthCollector;
import com.a.eye.skywalking.health.report.HeathReading;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.lmax.disruptor.EventHandler;
import java.util.ArrayList;
@ -14,25 +16,39 @@ import java.util.List;
* Created by wusheng on 2016/11/24.
*/
public class SendAckSpanEventHandler implements EventHandler<AckSpanHolder> {
private static ILog logger = LogManager.getLogger(SendAckSpanEventHandler.class);
private int bufferSize = 100;
private List<AckSpan> buffer = new ArrayList<>(bufferSize);
private static ILog logger = LogManager.getLogger(SendAckSpanEventHandler.class);
private int bufferSize = 100;
private AckSpan[] buffer = new AckSpan[bufferSize];
private int bufferIdx = 0;
public SendAckSpanEventHandler() {
Agent2RoutingClient.INSTANCE.setAckSpanDataSupplier(this);
}
@Override
public void onEvent(AckSpanHolder event, long sequence, boolean endOfBatch) throws Exception {
buffer.add(event.getData());
if(buffer[bufferIdx] != null){
return;
}
if (endOfBatch || buffer.size() == bufferSize) {
try {
//TODO， use GRPC to send
buffer[bufferIdx] = event.getData();
bufferIdx++;
HealthCollector.getCurrentHeathReading("SendAckSpanEventHandler").updateData(HeathReading.INFO, "%s messages were successful consumed .", buffer.size());
} finally {
buffer.clear();
}
if (bufferIdx == buffer.length) {
bufferIdx = 0;
}
if (endOfBatch) {
HealthCollector.getCurrentHeathReading("SendAckSpanEventHandler").updateData(HeathReading.INFO, "AckSpan messages were successful consumed .");
}
}
public List<AckSpan> getBufferData(){
List<AckSpan> data = new ArrayList<AckSpan>(bufferSize);
for (int i = 0; i < buffer.length; i++) {
data.add(buffer[i]);
buffer[i] = null;
}
return data;
}
}

37
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/disruptor/request/SendRequestSpanEventHandler.java
View File

@ -1,5 +1,6 @@
package com.a.eye.skywalking.disruptor.request;
import com.a.eye.skywalking.client.Agent2RoutingClient;
import com.a.eye.skywalking.health.report.HealthCollector;
import com.a.eye.skywalking.health.report.HeathReading;
import com.a.eye.skywalking.logging.api.ILog;
@ -14,25 +15,39 @@ import java.util.List;
* Created by wusheng on 2016/11/24.
*/
public class SendRequestSpanEventHandler implements EventHandler<RequestSpanHolder> {
private static ILog logger = LogManager.getLogger(SendRequestSpanEventHandler.class);
private static final int bufferSize = 100;
private List<RequestSpan> buffer = new ArrayList<>(bufferSize);
private static ILog logger = LogManager.getLogger(SendRequestSpanEventHandler.class);
private static final int bufferSize = 100;
private RequestSpan[] buffer = new RequestSpan[bufferSize];
private int bufferIdx = 0;
public SendRequestSpanEventHandler() {
Agent2RoutingClient.INSTANCE.setRequestSpanDataSupplier(this);
}
@Override
public void onEvent(RequestSpanHolder event, long sequence, boolean endOfBatch) throws Exception {
buffer.add(event.getData());
if(buffer[bufferIdx] != null){
return;
}
if (endOfBatch || buffer.size() == bufferSize) {
try {
//TODO， use GRPC to send
buffer[bufferIdx] = event.getData();
bufferIdx++;
HealthCollector.getCurrentHeathReading("SendRequestSpanEventHandler").updateData(HeathReading.INFO, "%s messages were successful consumed .", buffer.size());
} finally {
buffer.clear();
}
if(bufferIdx == buffer.length){
bufferIdx = 0;
}
if (endOfBatch) {
HealthCollector.getCurrentHeathReading("SendRequestSpanEventHandler").updateData(HeathReading.INFO, "Request Span messages were successful consumed .");
}
}
public List<RequestSpan> getBufferData(){
List<RequestSpan> data = new ArrayList<RequestSpan>(bufferSize);
for (int i = 0; i < buffer.length; i++) {
data.add(buffer[i]);
buffer[i] = null;
}
return data;
}
}

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/invoke/monitor/BaseInvokeMonitor.java
View File

@ -11,7 +11,7 @@ import com.a.eye.skywalking.model.Identification;
import com.a.eye.skywalking.model.Span;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.protocol.util.BuriedPointMachineUtil;
import com.a.eye.skywalking.util.BuriedPointMachineUtil;
import java.util.HashSet;
import java.util.Set;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/invoke/monitor/LocalMethodInvokeMonitor.java
View File

@ -6,7 +6,7 @@ import com.a.eye.skywalking.model.ContextData;
import com.a.eye.skywalking.model.EmptyContextData;
import com.a.eye.skywalking.model.Identification;
import com.a.eye.skywalking.model.Span;
import com.a.eye.skywalking.protocol.util.ContextGenerator;
import com.a.eye.skywalking.util.ContextGenerator;
public class LocalMethodInvokeMonitor extends BaseInvokeMonitor {

3
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/invoke/monitor/RPCClientInvokeMonitor.java
View File

@ -5,8 +5,7 @@ import com.a.eye.skywalking.context.CurrentThreadSpanStack;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.model.*;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.protocol.util.ContextGenerator;
import com.a.eye.skywalking.util.ContextGenerator;
public class RPCClientInvokeMonitor extends BaseInvokeMonitor {

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/invoke/monitor/RPCServerInvokeMonitor.java
View File

@ -6,7 +6,7 @@ import com.a.eye.skywalking.model.ContextData;
import com.a.eye.skywalking.model.Identification;
import com.a.eye.skywalking.model.Span;
import com.a.eye.skywalking.model.SpanType;
import com.a.eye.skywalking.protocol.util.ContextGenerator;
import com.a.eye.skywalking.util.ContextGenerator;
public class RPCServerInvokeMonitor extends BaseInvokeMonitor {

3
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/logging/EasyLogger.java
View File

@ -1,8 +1,7 @@
package com.a.eye.skywalking.logging;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.protocol.util.LoggingUtil;
import com.a.eye.skywalking.util.LoggingUtil;
import java.net.URLEncoder;
import java.text.SimpleDateFormat;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/logging/SyncFileWriter.java
View File

@ -2,7 +2,7 @@ package com.a.eye.skywalking.logging;
import com.a.eye.skywalking.conf.Config;
import com.a.eye.skywalking.protocol.util.LoggingUtil;
import com.a.eye.skywalking.util.LoggingUtil;
import java.io.File;
import java.io.FileNotFoundException;

5
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/model/Identification.java
View File

@ -1,10 +1,7 @@
package com.a.eye.skywalking.model;
import com.a.eye.skywalking.api.IBuriedPointType;
import com.a.eye.skywalking.protocol.util.StringUtil;
import java.util.HashMap;
import java.util.Map;
import com.a.eye.skywalking.util.StringUtil;
public class Identification {
private String viewPoint;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/plugin/AbstractClassEnhancePluginDefine.java
View File

@ -2,7 +2,7 @@ package com.a.eye.skywalking.plugin;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.protocol.util.StringUtil;
import com.a.eye.skywalking.util.StringUtil;
import net.bytebuddy.dynamic.DynamicType;
import net.bytebuddy.pool.TypePool.Resolution;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/plugin/PluginCfg.java
View File

@ -7,7 +7,7 @@ import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import com.a.eye.skywalking.protocol.util.StringUtil;
import com.a.eye.skywalking.util.StringUtil;
public class PluginCfg {
public final static PluginCfg CFG = new PluginCfg();

3
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/plugin/interceptor/enhance/ClassEnhancePluginDefine.java
View File

@ -4,13 +4,12 @@ import static net.bytebuddy.jar.asm.Opcodes.ACC_PRIVATE;
import static net.bytebuddy.matcher.ElementMatchers.any;
import static net.bytebuddy.matcher.ElementMatchers.not;
import com.a.eye.skywalking.logging.EasyLogger;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.plugin.AbstractClassEnhancePluginDefine;
import com.a.eye.skywalking.plugin.PluginException;
import com.a.eye.skywalking.plugin.interceptor.EnhancedClassInstanceContext;
import com.a.eye.skywalking.protocol.util.StringUtil;
import com.a.eye.skywalking.util.StringUtil;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.dynamic.DynamicType;
import net.bytebuddy.implementation.MethodDelegation;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/protocol/util/BuriedPointMachineUtil.java → skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/util/BuriedPointMachineUtil.java
View File

@ -1,4 +1,4 @@
package com.a.eye.skywalking.protocol.util;
package com.a.eye.skywalking.util;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/protocol/util/ContextGenerator.java → skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/util/ContextGenerator.java
View File

@ -1,4 +1,4 @@
package com.a.eye.skywalking.protocol.util;
package com.a.eye.skywalking.util;
import com.a.eye.skywalking.conf.Config;
import com.a.eye.skywalking.model.ContextData;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/protocol/util/LoggingUtil.java → skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/util/LoggingUtil.java
View File

@ -1,4 +1,4 @@
package com.a.eye.skywalking.protocol.util;
package com.a.eye.skywalking.util;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/protocol/util/StringUtil.java → skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/util/StringUtil.java
View File

@ -1,4 +1,4 @@
package com.a.eye.skywalking.protocol.util;
package com.a.eye.skywalking.util;
public final class StringUtil {
public static boolean isEmpty(String str) {

2
skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/protocol/util/TraceIdGenerator.java → skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/util/TraceIdGenerator.java
View File

@ -1,4 +1,4 @@
package com.a.eye.skywalking.protocol.util;
package com.a.eye.skywalking.util;
import java.util.UUID;

8
skywalking-storage-center/skywalking-storage/src/test/java/StorageThread.java
View File

@ -29,8 +29,8 @@ public class StorageThread extends Thread {
@Override
public void run() {
RequestSpan[] requestSpanList = new RequestSpan[10];
AckSpan[] ackSpanList = new AckSpan[10];
List<RequestSpan> requestSpanList = new ArrayList<RequestSpan>();
List<AckSpan> ackSpanList = new ArrayList<AckSpan>();
int cycle = 0;
for (int i = 0; i < count; i++) {
@ -55,8 +55,8 @@ public class StorageThread extends Thread {
listener.begin();
}
requestSpanList[cycle] = requestSpan;
ackSpanList[cycle] = ackSpan;
requestSpanList.add(requestSpan);
ackSpanList.add(ackSpan);
cycle++;
if (i % 10_000 == 0) {