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

解决部分代码编译问题

This commit is contained in:
ascrutae 2016-07-05 08:11:40 +08:00
parent b5e86d0330
commit 67e007a758
15 changed files with 172 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java
View File

@ -1,10 +1,8 @@
package com.ai.cloud.skywalking.buffer;
import com.ai.cloud.skywalking.conf.Config;
import com.ai.cloud.skywalking.conf.Constants;
import com.ai.cloud.skywalking.logging.LogManager;
import com.ai.cloud.skywalking.logging.Logger;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import com.ai.cloud.skywalking.selfexamination.HeathReading;
import com.ai.cloud.skywalking.selfexamination.SDKHealthCollector;
@ -45,7 +43,7 @@ public class BufferGroup {
}
}
public void save(Span span) {
public void save(ISerializable data) {
int i = index.getAndIncrement();
if (dataBuffer[i] != null) {
logger.warn(
@ -53,7 +51,7 @@ public class BufferGroup {
groupName, i);
SDKHealthCollector.getCurrentHeathReading("BufferGroup").updateData(HeathReading.WARNING, "BufferGroup index[" + i + "] data collision, data been coverd.");
}
dataBuffer[i] = span;
dataBuffer[i] = data;
SDKHealthCollector.getCurrentHeathReading("BufferGroup").updateData(HeathReading.INFO, "save span");
}
@ -69,7 +67,6 @@ public class BufferGroup {
@Override
public void run() {
//TODO: 根据list长度进行限制
List<ISerializable> packageData = new ArrayList<ISerializable>();
while (true) {
boolean bool = false;
@ -79,33 +76,33 @@ public class BufferGroup {
continue;
}
bool = true;
if (data.length() + dataBuffer[i].toString().length() >= Config.Sender.MAX_SEND_LENGTH) {
if (packageData.size() >= Config.Sender.MAX_SEND_DATA_SIZE) {
while (!DataSenderFactoryWithBalance.getSender()
.send(data.toString())) {
.send(packageData)) {
try {
Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL);
} catch (InterruptedException e) {
logger.error("Sleep Failure");
}
}
logger.debug("send buried-point data, size:{}", data.length());
data = new StringBuilder();
logger.debug("send buried-point data, size:{}", packageData.size());
packageData = new ArrayList<ISerializable>();
}
data.append(dataBuffer[i] + Constants.DATA_SPILT);
packageData.add(dataBuffer[i]);
dataBuffer[i] = null;
}
if (data != null && data.length() > 0) {
if (packageData != null && packageData.size() > 0) {
while (!DataSenderFactoryWithBalance.getSender().send(
data.toString())) {
packageData)) {
try {
Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL);
} catch (InterruptedException e) {
logger.error("Sleep Failure");
}
}
data = new StringBuilder();
packageData = new ArrayList<ISerializable>();
}
} catch (Throwable e) {
logger.error("buffer group running failed", e);

2
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java
View File

@ -46,6 +46,8 @@ public class Config {
}
public static class Sender {
// 最大发送数据个数
public static final int MAX_SEND_DATA_SIZE = 10;
// 最大发送者的连接数阀比例
public static int CONNECT_PERCENT = 50;

11
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java
View File

@ -1,5 +1,6 @@
package com.ai.cloud.skywalking.invoke.monitor;
import com.ai.cloud.skywalking.buffer.BufferGroup;
import com.ai.cloud.skywalking.buffer.ContextBuffer;
import com.ai.cloud.skywalking.conf.AuthDesc;
import com.ai.cloud.skywalking.conf.Config;
@ -7,6 +8,8 @@ import com.ai.cloud.skywalking.context.CurrentThreadSpanStack;
import com.ai.cloud.skywalking.logging.LogManager;
import com.ai.cloud.skywalking.logging.Logger;
import com.ai.cloud.skywalking.model.ContextData;
import com.ai.cloud.skywalking.protocol.AckSpan;
import com.ai.cloud.skywalking.protocol.RequestSpan;
import com.ai.cloud.skywalking.protocol.Span;
import java.util.HashSet;
@ -31,6 +34,8 @@ public abstract class BaseInvokeMonitor {
+ "\tParentLevelId:" + spanData.getParentLevel()
+ "\tLevelId:" + spanData.getLevelId());
}
// 根据SpanData生成RequestSpan，并保存
ContextBuffer.save(new RequestSpan(spanData));
// 将新创建的Context存放到ThreadLocal栈中。
CurrentThreadSpanStack.push(spanData);
@ -58,10 +63,10 @@ public abstract class BaseInvokeMonitor {
+ "\tviewpointId:" + spanData.getViewPointId()
+ "\tParentLevelId:" + spanData.getParentLevel()
+ "\tLevelId:" + spanData.getLevelId()
+ "\tbusinessKey:" + spanData.getBusinessKey());
+ "\tbusinessKey:" + spanData.getParameters());
}
ContextBuffer.save(spanData);
// 生成并保存到缓存
ContextBuffer.save(new AckSpan(spanData));
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}

3
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/RPCClientInvokeMonitor.java
View File

@ -1,5 +1,6 @@
package com.ai.cloud.skywalking.invoke.monitor;
import com.ai.cloud.skywalking.buffer.ContextBuffer;
import com.ai.cloud.skywalking.conf.AuthDesc;
import com.ai.cloud.skywalking.context.CurrentThreadSpanStack;
import com.ai.cloud.skywalking.logging.LogManager;
@ -7,6 +8,7 @@ import com.ai.cloud.skywalking.logging.Logger;
import com.ai.cloud.skywalking.model.ContextData;
import com.ai.cloud.skywalking.model.EmptyContextData;
import com.ai.cloud.skywalking.model.Identification;
import com.ai.cloud.skywalking.protocol.RequestSpan;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.cloud.skywalking.protocol.common.SpanType;
import com.ai.cloud.skywalking.util.ContextGenerator;
@ -25,6 +27,7 @@ public class RPCClientInvokeMonitor extends BaseInvokeMonitor {
//设置SpanType的类型
spanData.setSpanType(SpanType.RPC_CLIENT);
ContextBuffer.save(new RequestSpan(spanData));
CurrentThreadSpanStack.push(spanData);
return new ContextData(spanData.getTraceId(), generateSubParentLevelId(spanData), spanData.getCallType());

4
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java
View File

@ -70,7 +70,7 @@ public class DataSender implements IDataSender {
/**
* 返回是否发送成功
*
* @param data
* @param packageData
* @return
*/
@Override
@ -78,7 +78,7 @@ public class DataSender implements IDataSender {
try {
if (channel != null && channel.isActive()) {
byte[] dataPackage = TransportPackager.pack(data.getBytes());
byte[] dataPackage = TransportPackager.pack(packageData);
channel.writeAndFlush(dataPackage);
SDKHealthCollector.getCurrentHeathReading("sender").updateData(HeathReading.INFO, "DataSender[" + socketAddress + "] send data successfully.");

2
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java
View File

@ -53,7 +53,7 @@ public class DataSenderWithCopies implements IDataSender {
public boolean send(List<ISerializable> packageData) {
int successNum = 0;
for (IDataSender sender : senders) {
if (sender.send(data)) {
if (sender.send(packageData)) {
successNum++;
}
}

6
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/IDataSender.java
View File

@ -1,5 +1,9 @@
package com.ai.cloud.skywalking.sender;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import java.util.List;
public interface IDataSender {
public boolean send(String data);
public boolean send(List<ISerializable> data);
}

9
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java
View File

@ -37,6 +37,15 @@ public class AckSpan extends AbstractDataSerializable {
*/
private String exceptionStack;
public AckSpan(Span spanData) {
this.traceId = spanData.getTraceId();
this.parentLevel = spanData.getParentLevel();
this.levelId = spanData.getLevelId();
this.cost = System.currentTimeMillis() - spanData.getStartDate();
this.statusCode = spanData.getStatusCode();
this.exceptionStack = spanData.getExceptionStack();
}
public String getTraceId() {
return traceId;
}

19
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java
View File

@ -78,6 +78,25 @@ public class RequestSpan extends AbstractDataSerializable {
private Map<String, String> paramters = new HashMap<String, String>();
public RequestSpan(Span spanData) {
this.traceId = spanData.getTraceId();
this.parentLevel = spanData.getParentLevel();
this.levelId = spanData.getLevelId();
this.address = spanData.getAddress();
this.applicationId = spanData.getApplicationId();
this.callType = spanData.getCallType();
this.spanType = spanData.getSpanType();
this.spanTypeDesc = spanData.getSpanTypeDesc();
this.userId = spanData.getUserId();
if (isEntrySpan(spanData)) {
this.paramters.putAll(spanData.getParamters());
}
}
private boolean isEntrySpan(Span spanData) {
return "0".equals(spanData.getParentLevel() + spanData.getLevelId());
}
public String getTraceId() {
return traceId;
}

22
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializableDataTypeRegister.java
View File

@ -1,9 +1,13 @@
package com.ai.cloud.skywalking.protocol;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import com.ai.cloud.skywalking.protocol.common.NullableClass;
import com.ai.cloud.skywalking.protocol.exception.SerializableDataTypeRegisterException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
/**
* Created by wusheng on 16/7/4.
@ -33,6 +37,24 @@ public class SerializableDataTypeRegister {
}
}
public static ISerializable findSerializableClassAndSerialize(byte[] data) {
Set<Class<?>> registerSerializableClasses = CLASS_MAPPING_DATA_TYPE.keySet();
for (Class<?> serializableClass : registerSerializableClasses) {
try {
ISerializable result = ((ISerializable) serializableClass.newInstance());
//TODO
NullableClass nullableClass = result.convert2Object(data);
if (!nullableClass.isNull()){
}
} catch (Exception e) {
}
}
return null;
}
public static Integer getType(Class<?> clazz) {
if (CLASS_MAPPING_DATA_TYPE.containsKey(clazz)) {
return CLASS_MAPPING_DATA_TYPE.get(clazz);

19
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SpanData.java
View File

@ -3,6 +3,9 @@ package com.ai.cloud.skywalking.protocol;
import com.ai.cloud.skywalking.protocol.common.CallType;
import com.ai.cloud.skywalking.protocol.common.SpanType;
import java.util.HashMap;
import java.util.Map;
public abstract class SpanData {
/**
* Span在序列中，各字段间的分隔符
@ -92,7 +95,7 @@ public abstract class SpanData {
* 节点调用过程中的业务字段<br/>
* 如：业务系统设置的订单号，SQL语句等
*/
protected String businessKey = "";
protected Map<String, String> parameters = new HashMap<String, String>();
/**
* 节点调用的所在进程号
*/
@ -177,10 +180,6 @@ public abstract class SpanData {
return isInvalidate;
}
public void setBusinessKey(String businessKey) {
this.businessKey = businessKey;
}
public void setProcessNo(String processNo) {
this.processNo = processNo;
}
@ -209,8 +208,12 @@ public abstract class SpanData {
this.exceptionStack = exceptionStack;
}
public String getBusinessKey() {
return businessKey;
public Map<String, String> getParameters() {
return parameters;
}
public void setParameters(Map<String, String> parameters) {
this.parameters = parameters;
}
public String getProcessNo() {
@ -232,4 +235,6 @@ public abstract class SpanData {
public String getCallType() {
return callType;
}
}

2
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java
View File

@ -37,7 +37,7 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl
}
@Override
public Object convert2Object(byte[] data) {
public NullableClass convert2Object(byte[] data) {
// TODO:data的前4位转成type;
int dataType = 1;
if(!SerializableDataTypeRegister.isTypeAndClassMatch(dataType, this.getClass())){

2
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java
View File

@ -6,6 +6,6 @@ package com.ai.cloud.skywalking.protocol.common;
public interface ISerializable {
byte[] convert2Bytes();
Object convert2Object(byte[] data);
NullableClass convert2Object(byte[] data);
}

9
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/IntegerAssist.java
View File

@ -12,4 +12,13 @@ public class IntegerAssist {
src[3] = (byte) (value & 0xFF);
return src;
}
public static int bytesToInt(byte[] src, int offset) {
int value;
value = (int) (((src[offset] & 0xFF) << 24)
| ((src[offset + 1] & 0xFF) << 16)
| ((src[offset + 2] & 0xFF) << 8)
| (src[offset + 3] & 0xFF));
return value;
}
}

90
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java
View File

@ -1,48 +1,52 @@
package com.ai.cloud.skywalking.util;
import com.ai.cloud.skywalking.protocol.SerializableDataTypeRegister;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class TransportPackager {
public static byte[] pack(List<ISerializable> packageData) {
public static byte[] pack(List<ISerializable> beSendingData) {
// 对协议格式进行修改
// | check sum(4 byte) | data
byte[] dataPackage = new byte[data.length + 4];
packDataText(data, dataPackage);
packCheckSum(data, dataPackage);
byte[] dataText = packDataText(beSendingData);
byte[] dataPackage = packCheckSum(dataText);
return dataPackage;
}
private static void packCheckSum(byte[] data, byte[] dataPackage) {
byte[] checkSumArray = generateChecksum(data, 0);
System.arraycopy(checkSumArray, 0, dataPackage, 0, 4);
}
private static void packDataText(byte[] data, byte[] dataPackage) {
System.arraycopy(data, 0, dataPackage, 4, data.length);
}
public static List<ISerializable> unpack(byte[] dataPackage) {
if (validateCheckSum(dataPackage)) {
return unpackDataText(dataPackage);
} else {
return null;
}
}
private static byte[] unpackDataText(byte[] dataPackage) {
byte[] data = new byte[dataPackage.length - 4];
System.arraycopy(dataPackage, 4, data, 0, data.length);
return data;
private static List<ISerializable> unpackDataText(byte[] dataPackage) {
List<ISerializable> serializeData = new ArrayList<ISerializable>();
int currentLength = 0;
while (true) {
//读取长度
int dataLength = IntegerAssist.bytesToInt(dataPackage, 0);
// 反序列化
byte[] data = new byte[dataLength];
System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength);
//
ISerializable data1 = SerializableDataTypeRegister.findSerializableClassAndSerialize(data);
if (data1 != null) {
serializeData.add(data1);
}
currentLength = 4 + dataLength;
if (currentLength >= dataPackage.length) {
break;
}
}
return serializeData;
}
private static boolean validateCheckSum(byte[] dataPackage){
private static boolean validateCheckSum(byte[] dataPackage) {
byte[] checkSum = generateChecksum(dataPackage, 4);
byte[] originCheckSum = new byte[4];
System.arraycopy(dataPackage, 0, originCheckSum, 0, 4);
@ -63,4 +67,46 @@ public class TransportPackager {
return IntegerAssist.intToBytes(result);
}
private static byte[] packCheckSum(byte[] dataText) {
byte[] dataPackage = new byte[dataText.length + 4];
byte[] checkSum = generateChecksum(dataText, 0);
System.arraycopy(checkSum, 0, dataPackage, 0, 4);
System.arraycopy(dataText, 0, dataPackage, 4, dataText.length);
return dataPackage;
}
private static byte[] packDataText(List<ISerializable> beSendingData) {
byte[] dataText = new byte[1024 * 30];
int currentIndex = 0;
for (ISerializable sendingData : beSendingData) {
byte[] dataElementText = appendingLength(sendingData.convert2Bytes());
dataText = expansionCapacityIfNecessary(dataText, currentIndex, dataElementText);
appendBeSendingDataText(dataElementText, dataText, currentIndex);
currentIndex += dataElementText.length;
}
return dataText;
}
private static void appendBeSendingDataText(byte[] beSendingDataText, byte[] dataText, int currentIndex) {
System.arraycopy(beSendingDataText, 0, dataText, currentIndex, beSendingDataText.length);
}
private static byte[] expansionCapacityIfNecessary(byte[] dataText, int currentLength, byte[] beSendingDataText) {
if (beSendingDataText.length + currentLength > 1024 * 30) {
byte[] newDataText = new byte[dataText.length + 1024 * 30];
System.arraycopy(dataText, 0, newDataText, 0, dataText.length);
return newDataText;
}
return dataText;
}
private static byte[] appendingLength(byte[] dataByte) {
byte[] dataText = new byte[dataByte.length + 4];
System.arraycopy(dataByte, 0, dataText, 4, dataByte.length);
byte[] length = IntegerAssist.intToBytes(dataByte.length);
System.arraycopy(length, 0, dataText, 0, 4);
return dataText;
}
}