diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java index 1fa4634f4..574fb793c 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java @@ -5,11 +5,15 @@ 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; import com.ai.cloud.skywalking.sender.DataSenderFactoryWithBalance; import com.ai.cloud.skywalking.util.AtomicRangeInteger; +import java.util.ArrayList; +import java.util.List; + import static com.ai.cloud.skywalking.conf.Config.Buffer.BUFFER_MAX_SIZE; import static com.ai.cloud.skywalking.conf.Config.Consumer.*; @@ -17,7 +21,7 @@ public class BufferGroup { private static Logger logger = LogManager.getLogger(BufferGroup.class); private String groupName; //注意: 修改这个变量名,需要修改test-api工程的Config类中的SPAN_ARRAY_FIELD_NAME变量 - private Span[] dataBuffer = new Span[BUFFER_MAX_SIZE]; + private ISerializable[] dataBuffer = new ISerializable[BUFFER_MAX_SIZE]; AtomicRangeInteger index = new AtomicRangeInteger(0, BUFFER_MAX_SIZE); public BufferGroup(String groupName) { @@ -65,7 +69,8 @@ public class BufferGroup { @Override public void run() { - StringBuilder data = new StringBuilder(); + //TODO: 根据list长度进行限制 + List packageData = new ArrayList(); while (true) { boolean bool = false; try { diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/ContextBuffer.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/ContextBuffer.java index ad14db047..c4e1e6b2d 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/ContextBuffer.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/ContextBuffer.java @@ -3,6 +3,7 @@ package com.ai.cloud.skywalking.buffer; 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 java.util.concurrent.ThreadLocalRandom; @@ -18,9 +19,9 @@ public class ContextBuffer { //non } - public static void save(Span span) { + public static void save(ISerializable data) { try{ - pool.save(span); + pool.save(data); }catch(Throwable t){ logger.error("save span error.", t); } @@ -36,8 +37,8 @@ public class ContextBuffer { } } - public void save(Span span) { - bufferGroups[ThreadLocalRandom.current().nextInt(0, POOL_SIZE)].save(span); + public void save(ISerializable data) { + bufferGroups[ThreadLocalRandom.current().nextInt(0, POOL_SIZE)].save(data); } } diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java index 79c5128de..f1b56f3c3 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java @@ -1,5 +1,6 @@ package com.ai.cloud.skywalking.sender; +import com.ai.cloud.skywalking.protocol.common.ISerializable; import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; @@ -18,6 +19,7 @@ import io.netty.handler.codec.bytes.ByteArrayEncoder; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.List; import com.ai.cloud.skywalking.selfexamination.HeathReading; import com.ai.cloud.skywalking.selfexamination.SDKHealthCollector; @@ -72,7 +74,7 @@ public class DataSender implements IDataSender { * @return */ @Override - public boolean send(String data) { + public boolean send(List packageData) { try { if (channel != null && channel.isActive()) { diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java index c9ec369cd..c562ee002 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java @@ -3,8 +3,10 @@ package com.ai.cloud.skywalking.sender; import static com.ai.cloud.skywalking.conf.Config.Sender.MAX_COPY_NUM; import java.util.HashSet; +import java.util.List; import java.util.Set; +import com.ai.cloud.skywalking.protocol.common.ISerializable; import com.ai.cloud.skywalking.selfexamination.HeathReading; import com.ai.cloud.skywalking.selfexamination.SDKHealthCollector; @@ -48,7 +50,7 @@ public class DataSenderWithCopies implements IDataSender { /** * 尝试向所有副本发送 */ - public boolean send(String data) { + public boolean send(List packageData) { int successNum = 0; for (IDataSender sender : senders) { if (sender.send(data)) { diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java index 8fa9dbae6..26ef3bbdf 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java @@ -9,33 +9,33 @@ public class AckSpan extends AbstractDataSerializable { /** * tid,调用链的全局唯一标识 */ - protected String traceId; + private String traceId; /** * 当前调用链的上级描述
* 如当前序号为:0.1.0时,parentLevel=0.1 */ - protected String parentLevel; + private String parentLevel; /** * 当前调用链的本机描述
* 如当前序号为:0.1.0时,levelId=0 */ - protected int levelId = 0; + private int levelId = 0; /** * 节点调用花费时间 */ - protected long cost = 0L; + private long cost = 0L; /** * 节点调用的状态
* 0:成功
* 1:异常
* 异常判断原则:代码产生exception,并且此exception不在忽略列表中 */ - protected byte statusCode = 0; + private byte statusCode = 0; /** * 节点调用的错误堆栈
* 堆栈以JAVA的exception为主要判断依据 */ - protected String exceptionStack; + private String exceptionStack; public String getTraceId() { return traceId; diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java index 61f2983ff..70b6781e6 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java @@ -4,6 +4,9 @@ import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable; import com.ai.cloud.skywalking.protocol.common.CallType; import com.ai.cloud.skywalking.protocol.common.SpanType; +import java.util.HashMap; +import java.util.Map; + /** * Created by wusheng on 16/7/4. */ @@ -11,66 +14,69 @@ public class RequestSpan extends AbstractDataSerializable { /** * tid,调用链的全局唯一标识 */ - protected String traceId; + private String traceId; /** * 当前调用链的上级描述
* 如当前序号为:0.1.0时,parentLevel=0.1 */ - protected String parentLevel; + private String parentLevel; /** * 当前调用链的本机描述
* 如当前序号为:0.1.0时,levelId=0 */ - protected int levelId = 0; + private int levelId = 0; /** * 调用链中单个节点的入口描述
* 如:java方法名,调用的RPC地址等等 */ - protected String viewPointId = ""; + private String viewPointId = ""; /** * 节点调用开始时间 */ - protected long startDate = System.currentTimeMillis(); + private long startDate = System.currentTimeMillis(); /** * 节点调用的发生机器描述
* 包含机器名 + IP地址 */ - protected String address = ""; + private String address = ""; /** * 节点类型描述
* 已字符串的形式描述
* 如:java,dubbo等 */ - protected String spanTypeDesc = ""; + private String spanTypeDesc = ""; /** * 节点调用类型描述
+ * * @see CallType */ - protected String callType = ""; + private String callType = ""; /** * 节点分布式类型
* 本地调用 / RPC服务端 / RPC客户端 */ - protected SpanType spanType = SpanType.LOCAL; + private SpanType spanType = SpanType.LOCAL; /** * 节点调用的所在进程号 */ - protected String processNo = ""; + private String processNo = ""; /** * 节点调用所在的系统逻辑名称
* 由授权文件指定 */ - protected String applicationId = ""; + private String applicationId = ""; /** * 用户id
* 由授权文件指定 */ - protected String userId; + private String userId; + + private Map paramters = new HashMap(); public String getTraceId() { return traceId; @@ -168,6 +174,14 @@ public class RequestSpan extends AbstractDataSerializable { this.userId = userId; } + public Map getParamters() { + return paramters; + } + + public void setParamters(Map paramters) { + this.paramters = paramters; + } + @Override public int getDataType() { return 1; diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java index 4df3469fe..619d0bab8 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java @@ -21,11 +21,18 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl public abstract byte[] getData(); + /** + * 消息包结构: + * 4位消息体类型 + * n位数据正文 + * + * @return + */ @Override public byte[] convert2Bytes() { byte[] type = IntegerAssist.intToBytes(SerializableDataTypeRegister.getType(this.getClass())); - //TODO:类型+ data = 消息包 + //TODO:消息包 = 4位 return getData(); } diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java index 2eb326411..2a9ad6861 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java @@ -1,9 +1,12 @@ package com.ai.cloud.skywalking.util; +import com.ai.cloud.skywalking.protocol.common.ISerializable; + import java.util.Arrays; +import java.util.List; public class TransportPackager { - public static byte[] pack(byte[] data) { + public static byte[] pack(List packageData) { // 对协议格式进行修改 // | check sum(4 byte) | data byte[] dataPackage = new byte[data.length + 4]; @@ -24,7 +27,7 @@ public class TransportPackager { } - public static byte[] unpack(byte[] dataPackage) { + public static List unpack(byte[] dataPackage) { if (validateCheckSum(dataPackage)) { return unpackDataText(dataPackage); } else {