修复bug，提交测试类
This commit is contained in:
ascrutae
parent
81fa3b3ac3
commit
d26209006a
|
|
@ -30,7 +30,7 @@ public class RPCClientInvokeMonitor extends BaseInvokeMonitor {
|
|||
ContextBuffer.save(new RequestSpan(spanData));
|
||||
CurrentThreadSpanStack.push(spanData);
|
||||
|
||||
return new ContextData(spanData.getTraceId(), generateSubParentLevelId(spanData), spanData.getCallType());
|
||||
return new ContextData(spanData.getTraceId(), generateSubParentLevelId(spanData));
|
||||
} catch (Throwable t) {
|
||||
logger.error(t.getMessage(), t);
|
||||
return new EmptyContextData();
|
||||
|
|
|
|||
|
|
@ -27,18 +27,18 @@ public class RPCServerInvokeMonitor extends BaseInvokeMonitor {
|
|||
|
||||
invalidateAllSpanIfIsNotFirstSpan(spanData);
|
||||
|
||||
super.beforeInvoke(spanData);
|
||||
super.beforeInvoke(spanData, id);
|
||||
} catch (Throwable t) {
|
||||
logger.error(t.getMessage(), t);
|
||||
}
|
||||
}
|
||||
|
||||
public void afterInvoke(){
|
||||
public void afterInvoke() {
|
||||
super.afterInvoke();
|
||||
}
|
||||
|
||||
|
||||
public void occurException(Throwable th){
|
||||
public void occurException(Throwable th) {
|
||||
super.occurException(th);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -7,23 +7,20 @@ public class ContextData {
|
|||
private String traceId;
|
||||
private String parentLevel;
|
||||
private int levelId;
|
||||
private String spanType;
|
||||
|
||||
ContextData() {
|
||||
|
||||
}
|
||||
|
||||
public ContextData(String traceId, String parentLevel, String spanType) {
|
||||
public ContextData(String traceId, String parentLevel) {
|
||||
this.traceId = traceId;
|
||||
this.parentLevel = parentLevel;
|
||||
this.spanType = spanType;
|
||||
}
|
||||
|
||||
public ContextData(Span span) {
|
||||
this.traceId = span.getTraceId();
|
||||
this.parentLevel = span.getParentLevel();
|
||||
this.levelId = span.getLevelId();
|
||||
this.spanType = span.getSpanTypeDesc();
|
||||
}
|
||||
|
||||
public ContextData(String contextDataStr) {
|
||||
|
|
@ -35,7 +32,6 @@ public class ContextData {
|
|||
this.traceId = value[0];
|
||||
this.parentLevel = value[1].trim();
|
||||
this.levelId = Integer.valueOf(value[2]);
|
||||
this.spanType = value[3];
|
||||
|
||||
}
|
||||
|
||||
|
|
@ -51,10 +47,6 @@ public class ContextData {
|
|||
return levelId;
|
||||
}
|
||||
|
||||
public String getSpanType() {
|
||||
return spanType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
StringBuilder stringBuilder = new StringBuilder();
|
||||
|
|
@ -67,8 +59,6 @@ public class ContextData {
|
|||
}
|
||||
stringBuilder.append("-");
|
||||
stringBuilder.append(levelId);
|
||||
stringBuilder.append("-");
|
||||
stringBuilder.append(spanType);
|
||||
return stringBuilder.toString();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
package test.ai.cloud.serialize;
|
||||
|
||||
import com.ai.cloud.skywalking.buffer.ContextBuffer;
|
||||
import com.ai.cloud.skywalking.protocol.AckSpan;
|
||||
import com.ai.cloud.skywalking.protocol.RequestSpan;
|
||||
import com.ai.cloud.skywalking.protocol.Span;
|
||||
import com.ai.cloud.skywalking.protocol.common.CallType;
|
||||
import com.ai.cloud.skywalking.protocol.common.SpanType;
|
||||
|
||||
public class SerializeTest {
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
Span spandata = new Span("1.0b.1461060884539.7d6d06e.22489.1271.103", "", 0);
|
||||
spandata.setSpanType(SpanType.LOCAL);
|
||||
spandata.setStartDate(System.currentTimeMillis() - 1000 * 60);
|
||||
AckSpan requestSpan = new AckSpan(spandata);
|
||||
ContextBuffer.save(requestSpan);
|
||||
|
||||
Thread.sleep(60 * 1000 * 10);
|
||||
}
|
||||
}
|
||||
|
|
@ -24,11 +24,11 @@ public class AckSpan extends AbstractDataSerializable {
|
|||
* 当前调用链的本机描述<br/>
|
||||
* 如当前序号为：0.1.0时，levelId=0
|
||||
*/
|
||||
private int levelId = 0;
|
||||
private int levelId = 0;
|
||||
/**
|
||||
* 节点调用花费时间
|
||||
*/
|
||||
private long cost = 0L;
|
||||
private long cost = 0L;
|
||||
/**
|
||||
* 节点调用的状态<br/>
|
||||
* 0：成功<br/>
|
||||
|
|
@ -40,7 +40,7 @@ public class AckSpan extends AbstractDataSerializable {
|
|||
* 节点调用的错误堆栈<br/>
|
||||
* 堆栈以JAVA的exception为主要判断依据
|
||||
*/
|
||||
private String exceptionStack;
|
||||
private String exceptionStack = "";
|
||||
|
||||
|
||||
public AckSpan(Span spanData) {
|
||||
|
|
@ -124,7 +124,8 @@ public class AckSpan extends AbstractDataSerializable {
|
|||
|
||||
@Override
|
||||
public byte[] getData() {
|
||||
return new byte[0];
|
||||
return TraceProtocol.AckSpan.newBuilder().setTraceId(traceId).setParentLevel(parentLevel).
|
||||
setLevelId(levelId).setCost(cost).setStatusCode(statusCode).setExceptionStack(exceptionStack).build().toByteArray();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ public class Span {
|
|||
* 节点调用的错误堆栈<br/>
|
||||
* 堆栈以JAVA的exception为主要判断依据
|
||||
*/
|
||||
protected String exceptionStack;
|
||||
protected String exceptionStack = "";
|
||||
|
||||
/**
|
||||
* 节点的状态<br/>
|
||||
|
|
@ -60,7 +60,7 @@ public class Span {
|
|||
* 节点类型<br/>
|
||||
* 如：RPC Client,RPC Server,Local
|
||||
*/
|
||||
private SpanType spanType = SpanType.LOCAL;
|
||||
private SpanType spanType = SpanType.LOCAL;
|
||||
|
||||
public Span(String traceId) {
|
||||
this.traceId = traceId;
|
||||
|
|
|
|||
|
|
@ -28,13 +28,12 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl
|
|||
byte[] messageByteData = getData();
|
||||
byte[] messagePackage = new byte[4 + messageByteData.length];
|
||||
packData(messageByteData, messagePackage);
|
||||
setPackageLength(messagePackage);
|
||||
setProtocolType(messageByteData, messagePackage);
|
||||
return messagePackage;
|
||||
}
|
||||
|
||||
private void setPackageLength(byte[] messagePackage) {
|
||||
byte[] type = IntegerAssist.intToBytes(this.getDataType());
|
||||
System.arraycopy(type, 0, messagePackage, 0, type.length);
|
||||
private void setProtocolType(byte[] messageByteData, byte[] messagePackage) {
|
||||
System.arraycopy(IntegerAssist.intToBytes(getDataType()), 0, messagePackage, 0, 4);
|
||||
}
|
||||
|
||||
private void packData(byte[] messageByteData, byte[] messagePackage) {
|
||||
|
|
@ -49,7 +48,7 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl
|
|||
return new NullClass();
|
||||
}
|
||||
|
||||
return this.convertData(Arrays.copyOfRange(data,4, data.length));
|
||||
return this.convertData(Arrays.copyOfRange(data, 4, data.length));
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -29,7 +29,7 @@ public class SerializedFactory {
|
|||
if (abstractDataSerializable != null) {
|
||||
NullableClass nullableClass = abstractDataSerializable.convert2Object(bytes);
|
||||
if (!nullableClass.isNull()) {
|
||||
return abstractDataSerializable;
|
||||
return (AbstractDataSerializable) nullableClass;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
|
|
|
|||
|
|
@ -17,18 +17,22 @@ public class TransportPackager {
|
|||
|
||||
public static List<byte[]> unpack(byte[] dataPackage) {
|
||||
if (validateCheckSum(dataPackage)) {
|
||||
return unpackDataText(dataPackage);
|
||||
return unpackDataText(unpackCheckSum(dataPackage));
|
||||
} else {
|
||||
return new ArrayList<byte[]>();
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] unpackCheckSum(byte[] dataPackage) {
|
||||
return Arrays.copyOfRange(dataPackage, 4, dataPackage.length);
|
||||
}
|
||||
|
||||
private static List<byte[]> unpackDataText(byte[] dataPackage) {
|
||||
List<byte[]> serializeData = new ArrayList<byte[]>();
|
||||
int currentLength = 0;
|
||||
while (true) {
|
||||
//读取长度
|
||||
int dataLength = IntegerAssist.bytesToInt(dataPackage, 0);
|
||||
int dataLength = IntegerAssist.bytesToInt(dataPackage, currentLength);
|
||||
// 反序列化
|
||||
byte[] data = new byte[dataLength];
|
||||
System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength);
|
||||
|
|
@ -56,8 +60,8 @@ public class TransportPackager {
|
|||
* @return
|
||||
*/
|
||||
private static byte[] generateChecksum(byte[] data, int offset) {
|
||||
int result = data[0];
|
||||
for (int i = offset; i < data.length; i++) {
|
||||
int result = data[offset];
|
||||
for (int i = offset + 1; i < data.length; i++) {
|
||||
result ^= data[i];
|
||||
}
|
||||
|
||||
|
|
@ -73,12 +77,16 @@ public class TransportPackager {
|
|||
}
|
||||
|
||||
private static byte[] packDataText(List<ISerializable> beSendingData) {
|
||||
byte[] dataText = new byte[1024 * 30];
|
||||
byte[] dataText = null;
|
||||
int currentIndex = 0;
|
||||
for (ISerializable sendingData : beSendingData) {
|
||||
byte[] dataElementText = appendingLength(sendingData.convert2Bytes());
|
||||
dataText = expansionCapacityIfNecessary(dataText, currentIndex, dataElementText);
|
||||
appendBeSendingDataText(dataElementText, dataText, currentIndex);
|
||||
if (dataText == null) {
|
||||
dataText = new byte[dataElementText.length];
|
||||
} else {
|
||||
dataText = Arrays.copyOf(dataText, dataText.length + dataElementText.length);
|
||||
}
|
||||
System.arraycopy(dataElementText, 0, dataText, currentIndex, dataElementText.length);
|
||||
currentIndex += dataElementText.length;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
#采集服务器的端口
|
||||
server.port=34000
|
||||
server.max_deal_data_thread_number=5
|
||||
server.max_deal_data_thread_number=1
|
||||
server.failed_package_watching_time_windowss=300
|
||||
server.max_watching_failed_package_size=200;
|
||||
server.max_watching_failed_package_size=200
|
||||
|
||||
#每个线程最大缓存数量
|
||||
buffer.per_thread_max_buffer_number=1024
|
||||
|
|
@ -22,28 +22,8 @@ buffer.write_data_failure_retry_interval = 10000
|
|||
#数据包的最大限制
|
||||
datapackage.max_data_package=1048576
|
||||
|
||||
#定位文件时,每次读取偏移量跳过大小
|
||||
persistence.step_size_for_location_file_offset=20480
|
||||
#切换数据文件，等待时间(单位:毫秒)
|
||||
persistence.switch_file_wait_time=5000
|
||||
#追加EOF标志位的线程数量
|
||||
persistence.max_append_eof_flags_thread_number=2
|
||||
#当读取文件结束时最大等待时间
|
||||
persistence.read_ending_file_max_waite_time=50
|
||||
#每次存储的最大数量
|
||||
persistence.max_storage_size_per_time = 1048576
|
||||
|
||||
#偏移量注册文件的目录
|
||||
registerpersistence.register_file_parent_directory=d:/test-data/data/offset
|
||||
#偏移量注册文件名
|
||||
registerpersistence.register_file_name=offset.txt
|
||||
#偏移量注册备份文件名
|
||||
registerpersistence.register_bak_file_name=offset.txt.bak
|
||||
#偏移量写入文件等待周期(单位:毫秒)
|
||||
registerpersistence.offset_written_file_wait_cycle=5000
|
||||
|
||||
#hbase表名
|
||||
hbaseconfig.table_name=sw-call-chain
|
||||
hbaseconfig.table_name=sw-call-chain-new
|
||||
#hbase列簇名字
|
||||
hbaseconfig.family_column_name=call-chain
|
||||
#hbase zk quorum
|
||||
|
|
|
|||
Loading…
Reference in New Issue