diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/CurrentThreadSpanStack.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/CurrentThreadSpanStack.java
index 5e85daafb..251c803d0 100644
--- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/CurrentThreadSpanStack.java
+++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/CurrentThreadSpanStack.java
@@ -86,7 +86,7 @@ public class CurrentThreadSpanStack {
public void invalidatePresentSpans() {
for (SpanNode spanNode : spans) {
- spanNode.getData().setInvalidate(true);
+ spanNode.getData().setValidate(true);
}
}
}
diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java
index b5dacbb26..4c1e990eb 100644
--- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java
+++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java
@@ -31,12 +31,13 @@ public abstract class BaseInvokeMonitor {
logger.debug("TraceId:" + spanData.getTraceId() + "\tviewpointId:" + id.getViewPoint() + "\tParentLevelId:" + spanData.getParentLevel() + "\tLevelId:" + spanData
.getLevelId());
}
- // 根据SpanData生成RequestSpan，并保存
- ContextBuffer.save(RequestSpan.RequestSpanBuilder.
- newBuilder(spanData).callType(id.getCallType())
- .viewPoint(id.getViewPoint())
- .spanTypeDesc(id.getSpanTypeDesc())
- .build());
+
+
+ if (!spanData.isValidate()) {
+ // 根据SpanData生成RequestSpan，并保存
+ ContextBuffer.save(RequestSpan.RequestSpanBuilder.
+ newBuilder(spanData).callType(id.getCallType()).viewPoint(id.getViewPoint()).spanTypeDesc(id.getSpanTypeDesc()).build());
+ }
// 将新创建的Context存放到ThreadLocal栈中。
CurrentThreadSpanStack.push(spanData);
@@ -51,7 +52,7 @@ public abstract class BaseInvokeMonitor {
// 弹出上下文的栈顶中的元素
Span spanData = CurrentThreadSpanStack.pop();
- if (spanData == null || spanData.isInvalidate()) {
+ if (spanData == null || spanData.isValidate()) {
return;
}
diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/protocol/util/ContextGenerator.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/protocol/util/ContextGenerator.java
index b032ee6fe..54eadc97b 100644
--- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/protocol/util/ContextGenerator.java
+++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/protocol/util/ContextGenerator.java
@@ -63,7 +63,7 @@ public final class ContextGenerator {
// if true, current span is invalidate and current span also belong to RPC span
if (parentSpan.isRPCClientSpan()) {
span.setSpanType(SpanType.RPC_CLIENT);
- span.setIsInvalidate(true);
+ span.setValidate(false);
}
if (!StringUtil.isEmpty(parentSpan.getParentLevel())) {
diff --git a/skywalking-collector/skywalking-api/src/test/java/test/ai/cloud/serialize/SerializeTest.java b/skywalking-collector/skywalking-api/src/test/java/test/ai/cloud/serialize/SerializeTest.java
index ea3221941..9aecd1e6a 100644
--- a/skywalking-collector/skywalking-api/src/test/java/test/ai/cloud/serialize/SerializeTest.java
+++ b/skywalking-collector/skywalking-api/src/test/java/test/ai/cloud/serialize/SerializeTest.java
@@ -2,19 +2,19 @@ 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, "test-application", "test");
- spandata.setSpanType(SpanType.LOCAL);
- spandata.setStartDate(System.currentTimeMillis() - 1000 * 60);
- AckSpan requestSpan = new AckSpan(spandata);
- ContextBuffer.save(requestSpan);
+ while (true) {
+ Span spandata = new Span("1.0b.1461060884539.7d6d06e.22489.1271.103", "", 0, "test-application", "test");
+ spandata.setSpanType(SpanType.LOCAL);
+ spandata.setStartDate(System.currentTimeMillis() - 1000 * 60);
+ AckSpan requestSpan = new AckSpan(spandata);
+ ContextBuffer.save(requestSpan);
+ Thread.sleep(2 * 1000);
+ }
- Thread.sleep(60 * 1000 * 10);
}
}
diff --git a/skywalking-collector/skywalking-protocol/src/main/gen-java/com/ai/cloud/skywalking/protocol/proto/TraceProtocol.java b/skywalking-collector/skywalking-protocol/src/main/gen-java/com/ai/cloud/skywalking/protocol/proto/TraceProtocol.java
index 1261262b5..af9be17c9 100644
--- a/skywalking-collector/skywalking-protocol/src/main/gen-java/com/ai/cloud/skywalking/protocol/proto/TraceProtocol.java
+++ b/skywalking-collector/skywalking-protocol/src/main/gen-java/com/ai/cloud/skywalking/protocol/proto/TraceProtocol.java
@@ -1206,6 +1206,20 @@ public final class TraceProtocol {
*/
com.google.protobuf.ByteString
getBussinessKeyBytes();
+
+ /**
+ * required string agentId = 12;
+ */
+ boolean hasAgentId();
+ /**
+ * required string agentId = 12;
+ */
+ java.lang.String getAgentId();
+ /**
+ * required string agentId = 12;
+ */
+ com.google.protobuf.ByteString
+ getAgentIdBytes();
}
/**
* Protobuf type {@code RequestSpan}
@@ -1322,6 +1336,12 @@ public final class TraceProtocol {
bussinessKey_ = bs;
break;
}
+ case 98: {
+ com.google.protobuf.ByteString bs = input.readBytes();
+ bitField0_ |= 0x00000800;
+ agentId_ = bs;
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -1743,6 +1763,48 @@ public final class TraceProtocol {
}
}
+ public static final int AGENTID_FIELD_NUMBER = 12;
+ private java.lang.Object agentId_;
+ /**
+ * required string agentId = 12;
+ */
+ public boolean hasAgentId() {
+ return ((bitField0_ & 0x00000800) == 0x00000800);
+ }
+ /**
+ * required string agentId = 12;
+ */
+ public java.lang.String getAgentId() {
+ java.lang.Object ref = agentId_;
+ if (ref instanceof java.lang.String) {
+ return (java.lang.String) ref;
+ } else {
+ com.google.protobuf.ByteString bs =
+ (com.google.protobuf.ByteString) ref;
+ java.lang.String s = bs.toStringUtf8();
+ if (bs.isValidUtf8()) {
+ agentId_ = s;
+ }
+ return s;
+ }
+ }
+ /**
+ * required string agentId = 12;
+ */
+ public com.google.protobuf.ByteString
+ getAgentIdBytes() {
+ java.lang.Object ref = agentId_;
+ if (ref instanceof java.lang.String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ agentId_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+
private void initFields() {
traceId_ = "";
parentLevel_ = "";
@@ -1755,6 +1817,7 @@ public final class TraceProtocol {
applicationId_ = "";
userId_ = "";
bussinessKey_ = "";
+ agentId_ = "";
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -1798,6 +1861,10 @@ public final class TraceProtocol {
memoizedIsInitialized = 0;
return false;
}
+ if (!hasAgentId()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
memoizedIsInitialized = 1;
return true;
}
@@ -1838,6 +1905,9 @@ public final class TraceProtocol {
if (((bitField0_ & 0x00000400) == 0x00000400)) {
output.writeBytes(11, getBussinessKeyBytes());
}
+ if (((bitField0_ & 0x00000800) == 0x00000800)) {
+ output.writeBytes(12, getAgentIdBytes());
+ }
getUnknownFields().writeTo(output);
}
@@ -1891,6 +1961,10 @@ public final class TraceProtocol {
size += com.google.protobuf.CodedOutputStream
.computeBytesSize(11, getBussinessKeyBytes());
}
+ if (((bitField0_ & 0x00000800) == 0x00000800)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(12, getAgentIdBytes());
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -2030,6 +2104,8 @@ public final class TraceProtocol {
bitField0_ = (bitField0_ & ~0x00000200);
bussinessKey_ = "";
bitField0_ = (bitField0_ & ~0x00000400);
+ agentId_ = "";
+ bitField0_ = (bitField0_ & ~0x00000800);
return this;
}
@@ -2102,6 +2178,10 @@ public final class TraceProtocol {
to_bitField0_ |= 0x00000400;
}
result.bussinessKey_ = bussinessKey_;
+ if (((from_bitField0_ & 0x00000800) == 0x00000800)) {
+ to_bitField0_ |= 0x00000800;
+ }
+ result.agentId_ = agentId_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -2167,6 +2247,11 @@ public final class TraceProtocol {
bussinessKey_ = other.bussinessKey_;
onChanged();
}
+ if (other.hasAgentId()) {
+ bitField0_ |= 0x00000800;
+ agentId_ = other.agentId_;
+ onChanged();
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -2208,6 +2293,10 @@ public final class TraceProtocol {
return false;
}
+ if (!hasAgentId()) {
+
+ return false;
+ }
return true;
}
@@ -2934,6 +3023,82 @@ public final class TraceProtocol {
return this;
}
+ private java.lang.Object agentId_ = "";
+ /**
+ * required string agentId = 12;
+ */
+ public boolean hasAgentId() {
+ return ((bitField0_ & 0x00000800) == 0x00000800);
+ }
+ /**
+ * required string agentId = 12;
+ */
+ public java.lang.String getAgentId() {
+ java.lang.Object ref = agentId_;
+ if (!(ref instanceof java.lang.String)) {
+ com.google.protobuf.ByteString bs =
+ (com.google.protobuf.ByteString) ref;
+ java.lang.String s = bs.toStringUtf8();
+ if (bs.isValidUtf8()) {
+ agentId_ = s;
+ }
+ return s;
+ } else {
+ return (java.lang.String) ref;
+ }
+ }
+ /**
+ * required string agentId = 12;
+ */
+ public com.google.protobuf.ByteString
+ getAgentIdBytes() {
+ java.lang.Object ref = agentId_;
+ if (ref instanceof String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8(
+ (java.lang.String) ref);
+ agentId_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+ /**
+ * required string agentId = 12;
+ */
+ public Builder setAgentId(
+ java.lang.String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000800;
+ agentId_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * required string agentId = 12;
+ */
+ public Builder clearAgentId() {
+ bitField0_ = (bitField0_ & ~0x00000800);
+ agentId_ = getDefaultInstance().getAgentId();
+ onChanged();
+ return this;
+ }
+ /**
+ * required string agentId = 12;
+ */
+ public Builder setAgentIdBytes(
+ com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000800;
+ agentId_ = value;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:RequestSpan)
}
@@ -2967,14 +3132,14 @@ public final class TraceProtocol {
"\n\023TraceProtocol.proto\"z\n\007AckSpan\022\017\n\007trac" +
"eId\030\001 \002(\t\022\023\n\013parentLevel\030\002 \001(\t\022\017\n\007levelI" +
"d\030\003 \002(\005\022\014\n\004cost\030\004 \002(\003\022\022\n\nstatusCode\030\005 \002(" +
- "\005\022\026\n\016exceptionStack\030\006 \001(\t\"\343\001\n\013RequestSpa" +
+ "\005\022\026\n\016exceptionStack\030\006 \001(\t\"\364\001\n\013RequestSpa" +
"n\022\017\n\007traceId\030\001 \002(\t\022\023\n\013parentLevel\030\002 \001(\t\022" +
"\017\n\007levelId\030\003 \002(\005\022\023\n\013viewPointId\030\004 \002(\t\022\021\n" +
"\tstartDate\030\005 \002(\003\022\024\n\014spanTypeDesc\030\006 \002(\t\022\020" +
"\n\010callType\030\007 \002(\t\022\020\n\010spanType\030\010 \002(\r\022\025\n\rap" +
"plicationId\030\t \002(\t\022\016\n\006userId\030\n \002(\t\022\024\n\014bus" +
- "sinessKey\030\013 \001(\tB(\n&com.ai.cloud.skywalki",
- "ng.protocol.proto"
+ "sinessKey\030\013 \001(\t\022\017\n\007agentId\030\014 \002(\tB(\n&com.",
+ "ai.cloud.skywalking.protocol.proto"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor. InternalDescriptorAssigner() {
@@ -2999,7 +3164,7 @@ public final class TraceProtocol {
internal_static_RequestSpan_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_RequestSpan_descriptor,
- new java.lang.String[] { "TraceId", "ParentLevel", "LevelId", "ViewPointId", "StartDate", "SpanTypeDesc", "CallType", "SpanType", "ApplicationId", "UserId", "BussinessKey", });
+ new java.lang.String[] { "TraceId", "ParentLevel", "LevelId", "ViewPointId", "StartDate", "SpanTypeDesc", "CallType", "SpanType", "ApplicationId", "UserId", "BussinessKey", "AgentId", });
}
// @@protoc_insertion_point(outer_class_scope)
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 b28c64abf..ebf55c33e 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
@@ -1,6 +1,7 @@
package com.ai.cloud.skywalking.protocol;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
+import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
import com.ai.cloud.skywalking.protocol.proto.TraceProtocol;
import com.google.protobuf.InvalidProtocolBufferException;
@@ -24,18 +25,18 @@ public class AckSpan extends AbstractDataSerializable {
* 当前调用链的本机描述
* 如当前序号为：0.1.0时，levelId=0
*/
- private int levelId = 0;
+ private int levelId = 0;
/**
* 节点调用花费时间
*/
- private long cost = 0L;
+ private long cost = 0L;
/**
* 节点调用的状态
* 0：成功
* 1：异常
* 异常判断原则：代码产生exception，并且此exception不在忽略列表中
*/
- private byte statusCode = 0;
+ private byte statusCode = 0;
/**
* 节点调用的错误堆栈
* 堆栈以JAVA的exception为主要判断依据
@@ -124,11 +125,12 @@ public class AckSpan extends AbstractDataSerializable {
@Override
public byte[] getData() {
return TraceProtocol.AckSpan.newBuilder().setTraceId(traceId).setParentLevel(parentLevel).
- setLevelId(levelId).setCost(cost).setStatusCode(statusCode).setExceptionStack(exceptionStack).build().toByteArray();
+ setLevelId(levelId).setCost(cost).setStatusCode(statusCode).setExceptionStack(exceptionStack).build()
+ .toByteArray();
}
@Override
- public AbstractDataSerializable convertData(byte[] data) {
+ public AbstractDataSerializable convertData(byte[] data) throws ConvertFailedException {
AckSpan ackSpan = new AckSpan();
try {
TraceProtocol.AckSpan ackSpanProtocol = TraceProtocol.AckSpan.parseFrom(data);
@@ -139,7 +141,7 @@ public class AckSpan extends AbstractDataSerializable {
ackSpan.setExceptionStack(ackSpanProtocol.getExceptionStack());
ackSpan.setStatusCode((byte) ackSpanProtocol.getStatusCode());
} catch (InvalidProtocolBufferException e) {
- return null;
+ throw new ConvertFailedException();
}
return ackSpan;
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 38641eae7..5f4b903a9 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
@@ -3,6 +3,7 @@ package com.ai.cloud.skywalking.protocol;
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 com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
import com.ai.cloud.skywalking.protocol.proto.TraceProtocol;
import com.google.protobuf.InvalidProtocolBufferException;
@@ -79,6 +80,11 @@ public class RequestSpan extends AbstractDataSerializable {
*/
private String businessKey = "";
+ /**
+ * 实例ID
+ */
+ private String agentId = "";
+
public RequestSpan(Span spanData) {
this.traceId = spanData.getTraceId();
this.parentLevel = spanData.getParentLevel();
@@ -191,13 +197,15 @@ public class RequestSpan extends AbstractDataSerializable {
@Override
public byte[] getData() {
- return TraceProtocol.RequestSpan.newBuilder().setTraceId(traceId).setParentLevel(parentLevel).setLevelId(levelId).setViewPointId(viewPointId).setStartDate(startDate)
- .setSpanType(spanType.getValue()).setSpanTypeDesc(spanTypeDesc).setBussinessKey(businessKey).setCallType(callType).setApplicationId(applicationId).setUserId(userId)
- .build().toByteArray();
+ return TraceProtocol.RequestSpan.newBuilder().setTraceId(traceId).setParentLevel(parentLevel)
+ .setLevelId(levelId).setViewPointId(viewPointId).setStartDate(startDate)
+ .setSpanType(spanType.getValue()).setSpanTypeDesc(spanTypeDesc).setBussinessKey(businessKey)
+ .setCallType(callType).setApplicationId(applicationId).setUserId(userId).setBussinessKey(businessKey)
+ .setAgentId(agentId).build().toByteArray();
}
@Override
- public AbstractDataSerializable convertData(byte[] data) {
+ public AbstractDataSerializable convertData(byte[] data) throws ConvertFailedException {
RequestSpan requestSpan = new RequestSpan();
try {
TraceProtocol.RequestSpan requestSpanByte = TraceProtocol.RequestSpan.parseFrom(data);
@@ -211,8 +219,10 @@ public class RequestSpan extends AbstractDataSerializable {
requestSpan.setStartDate(requestSpanByte.getStartDate());
requestSpan.setUserId(requestSpanByte.getUserId());
requestSpan.setViewPointId(requestSpanByte.getViewPointId());
+ requestSpan.setBusinessKey(requestSpanByte.getBussinessKey());
+ requestSpan.setAgentId(requestSpanByte.getAgentId());
} catch (InvalidProtocolBufferException e) {
- return null;
+ throw new ConvertFailedException();
}
return requestSpan;
@@ -223,6 +233,17 @@ public class RequestSpan extends AbstractDataSerializable {
return false;
}
+ public void setBusinessKey(String businessKey) {
+ this.businessKey = businessKey;
+ }
+
+ public String getAgentId() {
+ return agentId;
+ }
+
+ public void setAgentId(String agentId) {
+ this.agentId = agentId;
+ }
public static class RequestSpanBuilder {
private RequestSpan ackSpan;
diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializedFactory.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializedFactory.java
index 62faaf672..8fd2d7f25 100644
--- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializedFactory.java
+++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializedFactory.java
@@ -2,6 +2,7 @@ package com.ai.cloud.skywalking.protocol;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.common.NullableClass;
+import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
import java.util.HashMap;
@@ -9,7 +10,8 @@ import java.util.Map;
import java.util.ServiceLoader;
public class SerializedFactory {
- public static Map serializableMap = new HashMap();
+ public static Map serializableMap =
+ new HashMap();
static {
@@ -20,13 +22,17 @@ public class SerializedFactory {
}
}
- public static AbstractDataSerializable unSerialize(byte[] bytes) {
- AbstractDataSerializable abstractDataSerializable = serializableMap.get(IntegerAssist.bytesToInt(bytes, 0));
- if (abstractDataSerializable != null) {
- NullableClass nullableClass = abstractDataSerializable.convert2Object(bytes);
- if (!nullableClass.isNull()) {
- return (AbstractDataSerializable) nullableClass;
+ public static AbstractDataSerializable unSerialize(byte[] bytes) throws ConvertFailedException {
+ try {
+ AbstractDataSerializable abstractDataSerializable = serializableMap.get(IntegerAssist.bytesToInt(bytes, 0));
+ if (abstractDataSerializable != null) {
+ NullableClass nullableClass = abstractDataSerializable.convert2Object(bytes);
+ if (!nullableClass.isNull()) {
+ return (AbstractDataSerializable) nullableClass;
+ }
}
+ } catch (Exception e) {
+ throw new ConvertFailedException();
}
return null;
}
diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java
index 7c29ccbac..f8b71aa23 100644
--- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java
+++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java
@@ -49,7 +49,7 @@ public class Span {
* 节点的状态
* 不参与序列化
*/
- protected boolean isInvalidate = false;
+ protected boolean isValidate = true;
/**
* 节点调用过程中的业务字段
@@ -111,8 +111,8 @@ public class Span {
this.startDate = startDate;
}
- public boolean isInvalidate() {
- return isInvalidate;
+ public boolean isValidate() {
+ return isValidate;
}
public byte getStatusCode() {
@@ -135,8 +135,8 @@ public class Span {
this.parameters = parameters;
}
- public void setInvalidate(boolean invalidate) {
- isInvalidate = invalidate;
+ public void setValidate(boolean validate) {
+ isValidate = validate;
}
public boolean isRPCClientSpan() {
@@ -154,10 +154,6 @@ public class Span {
return spanType;
}
- public void setIsInvalidate(boolean isInvalidate) {
- this.isInvalidate = isInvalidate;
- }
-
public void handleException(Throwable e, Set exclusiveExceptionSet, int maxExceptionStackLength) {
ByteArrayOutputStream buf = null;
StringBuilder expMessage = new StringBuilder();
@@ -217,4 +213,6 @@ public class Span {
public void setUserId(String userId) {
this.userId = userId;
}
+
+
}
diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/TransportPackager.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/TransportPackager.java
index 83e721743..c3eac035f 100644
--- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/TransportPackager.java
+++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/TransportPackager.java
@@ -16,44 +16,30 @@ public class TransportPackager {
return dataPackage;
}
- public static List unpack(byte[] dataPackage) {
- if (validateCheckSum(dataPackage)) {
- return unpackDataBody(unpackCheckSum(dataPackage));
- } else {
- return new ArrayList();
- }
- }
-
- private static byte[] unpackCheckSum(byte[] dataPackage) {
- return Arrays.copyOfRange(dataPackage, 4, dataPackage.length);
- }
-
- private static List unpackDataBody(byte[] dataPackage) {
- List serializeData = new ArrayList();
- int currentLength = 0;
- while (true) {
- //读取长度
- int dataLength = IntegerAssist.bytesToInt(dataPackage, currentLength);
- // 反序列化
- byte[] data = new byte[dataLength];
- System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength);
- //
- serializeData.add(data);
- currentLength = 4 + dataLength;
- if (currentLength >= dataPackage.length) {
- break;
+ public static List unpackDataBody(byte[] dataPackage) {
+ List serializeData = null;
+ try {
+ serializeData = new ArrayList();
+ int currentLength = 0;
+ while (true) {
+ //读取长度
+ int dataLength = IntegerAssist.bytesToInt(dataPackage, currentLength);
+ // 反序列化
+ byte[] data = new byte[dataLength];
+ System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength);
+ //
+ serializeData.add(data);
+ currentLength += 4 + dataLength;
+ if (currentLength >= dataPackage.length) {
+ break;
+ }
}
+ } catch (Exception e) {
+ e.printStackTrace();
}
return serializeData;
}
- private static boolean validateCheckSum(byte[] dataPackage) {
- byte[] checkSum = generateChecksum(dataPackage, 4);
- byte[] originCheckSum = new byte[4];
- System.arraycopy(dataPackage, 0, originCheckSum, 0, 4);
- return Arrays.equals(checkSum, originCheckSum);
- }
-
/**
* 生成校验和参数
*
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 155927931..c573fe20d 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
@@ -2,6 +2,7 @@ package com.ai.cloud.skywalking.protocol.common;
import com.ai.cloud.skywalking.protocol.NullClass;
import com.ai.cloud.skywalking.protocol.SerializedFactory;
+import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
import java.util.Arrays;
@@ -14,7 +15,7 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl
public abstract byte[] getData();
- public abstract AbstractDataSerializable convertData(byte[] data);
+ public abstract AbstractDataSerializable convertData(byte[] data) throws ConvertFailedException;
/**
* 消息包结构:
@@ -41,7 +42,7 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl
}
@Override
- public NullableClass convert2Object(byte[] data) {
+ public NullableClass convert2Object(byte[] data) throws ConvertFailedException {
int dataType = IntegerAssist.bytesToInt(data, 0);
if (!SerializedFactory.isCanSerialized(dataType)) {
diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java
index 81ce0d536..97cfe832e 100644
--- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java
+++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java
@@ -1,11 +1,13 @@
package com.ai.cloud.skywalking.protocol.common;
+import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
+
/**
* Created by wusheng on 16/7/4.
*/
public interface ISerializable {
byte[] convert2Bytes();
- NullableClass convert2Object(byte[] data);
+ NullableClass convert2Object(byte[] data) throws ConvertFailedException;
}
diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/exception/ConvertFailedException.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/exception/ConvertFailedException.java
new file mode 100644
index 000000000..63bd6a44d
--- /dev/null
+++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/exception/ConvertFailedException.java
@@ -0,0 +1,4 @@
+package com.ai.cloud.skywalking.protocol.exception;
+
+public class ConvertFailedException extends Exception {
+}
diff --git a/skywalking-collector/skywalking-protocol/src/main/proto/TraceProtocol.proto b/skywalking-collector/skywalking-protocol/src/main/proto/TraceProtocol.proto
index 214c3f6f8..77b490abd 100644
--- a/skywalking-collector/skywalking-protocol/src/main/proto/TraceProtocol.proto
+++ b/skywalking-collector/skywalking-protocol/src/main/proto/TraceProtocol.proto
@@ -1,8 +1,8 @@
-syntax="proto2";
+syntax = "proto2";
option java_package = "com.ai.cloud.skywalking.protocol.proto";
-message AckSpan{
+message AckSpan {
required string traceId = 1;
optional string parentLevel = 2;
required int32 levelId = 3;
@@ -24,4 +24,5 @@ message RequestSpan {
required string applicationId = 9;
required string userId = 10;
optional string bussinessKey = 11;
+ required string agentId = 12;
}
diff --git a/skywalking-collector/skywalking-sdk-plugin/dubbo-plugin/pom.xml b/skywalking-collector/skywalking-sdk-plugin/dubbo-plugin/pom.xml
index c4ff7e408..1e5460750 100644
--- a/skywalking-collector/skywalking-sdk-plugin/dubbo-plugin/pom.xml
+++ b/skywalking-collector/skywalking-sdk-plugin/dubbo-plugin/pom.xml
@@ -18,11 +18,6 @@
-
- com.ai.cloud
- skywalking-api
- ${project.version}
-
com.ai.cloud
skywalking-api
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java
index ce4187873..cee442c58 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java
@@ -4,6 +4,7 @@ import com.ai.cloud.skywalking.reciever.buffer.DataBufferThreadContainer;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.conf.ConfigInitializer;
import com.ai.cloud.skywalking.reciever.handler.CollectionServerDataHandler;
+import com.ai.cloud.skywalking.reciever.peresistent.PersistenceThreadLauncher;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
@@ -35,7 +36,8 @@ public class CollectionServer {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
- b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100).handler(new LoggingHandler(LogLevel.INFO))
+ b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100)
+ .handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer() {
@Override
public void initChannel(io.netty.channel.socket.SocketChannel ch) throws Exception {
@@ -61,6 +63,8 @@ public class CollectionServer {
initializeParam();
logger.info("To init server health collector...");
ServerHealthCollector.init();
+ logger.info("To launch register persistence thread....");
+ PersistenceThreadLauncher.doLaunch();
logger.info("To init data buffer thread container...");
DataBufferThreadContainer.init();
logger.info("Starting collection server.....");
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/AppendEOFFlagThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/AppendEOFFlagThread.java
new file mode 100644
index 000000000..c9a6d1020
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/AppendEOFFlagThread.java
@@ -0,0 +1,44 @@
+package com.ai.cloud.skywalking.reciever.buffer;
+
+import com.ai.cloud.skywalking.reciever.model.BufferDataPackagerGenerator;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+
+class AppendEOFFlagThread extends Thread {
+ private Logger logger = LogManager.getLogger(AppendEOFFlagThread.class);
+ private File[] dataBufferFiles;
+ private CountDownLatch countDownLatch;
+
+ public AppendEOFFlagThread(File[] dataBufferFiles, CountDownLatch countDownLatch) {
+ super("AppendEOFFlagThread");
+ this.dataBufferFiles = dataBufferFiles;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public void run() {
+ FileOutputStream fileOutputStream = null;
+ for (File file : dataBufferFiles) {
+ try {
+ logger.info("Add EOF flags to unprocessed data file[{}]", file.getName());
+ fileOutputStream = new FileOutputStream(new File(file.getParent(), file.getName()), true);
+ fileOutputStream.write(BufferDataPackagerGenerator.generateEOFPackage());
+ } catch (IOException e) {
+ logger.info("Add EOF flags to the unprocessed data file failed.", e);
+ } finally {
+ try {
+ fileOutputStream.flush();
+ fileOutputStream.close();
+ } catch (IOException e) {
+ logger.error("Flush data file failed", e);
+ }
+ }
+ countDownLatch.countDown();
+ }
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java
index 05bde4ee7..26f41e02b 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java
@@ -1,20 +1,18 @@
package com.ai.cloud.skywalking.reciever.buffer;
-import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
+import com.ai.cloud.skywalking.protocol.util.AtomicRangeInteger;
import com.ai.cloud.skywalking.reciever.conf.Config;
-import com.ai.cloud.skywalking.reciever.processor.IProcessor;
-import com.ai.cloud.skywalking.reciever.processor.ProcessorFactory;
+import com.ai.cloud.skywalking.reciever.model.BufferDataPackagerGenerator;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
-import com.ai.cloud.skywalking.protocol.SerializedFactory;
-import com.ai.cloud.skywalking.protocol.util.AtomicRangeInteger;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.UUID;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.DATA_CONFLICT_WAIT_TIME;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.PER_THREAD_MAX_BUFFER_NUMBER;
@@ -31,28 +29,30 @@ public class DataBufferThread extends Thread {
@Override
public void run() {
- Map> serializeObjects;
+ FileOutputStream fileOutputStream = null;
+ int length = 0;
while (true) {
- serializeObjects = new HashMap>();
for (int i = 0; i < data.length; i++) {
if (data[i] == null) {
continue;
}
- AbstractDataSerializable serializeData = SerializedFactory.unSerialize(data[i]);
- List hasBeenSerializedObjects = serializeObjects.get(serializeData.getDataType());
- if (hasBeenSerializedObjects == null) {
- serializeObjects.put(serializeData.getDataType(), new ArrayList());
+ if (fileOutputStream == null) {
+ fileOutputStream = acquiredNewBufferFileStream();
}
- serializeObjects.get(serializeData.getDataType()).add(serializeData);
- data[i] = null;
- }
+ try {
+ fileOutputStream.write(BufferDataPackagerGenerator.pack(data[i]));
+ length += data[i].length;
+ data[i] = null;
+ } catch (IOException e) {
+ logger.error("Failed to write msg.", e);
+ }
- for (Map.Entry> entry : serializeObjects.entrySet()) {
- IProcessor processor = ProcessorFactory.chooseProcessor(entry.getKey());
- if (processor != null) {
- processor.process(entry.getValue());
+
+ if (length > Config.Buffer.BUFFER_FILE_MAX_LENGTH) {
+ closeCurrentBufferFile(fileOutputStream);
+ fileOutputStream = null;
}
}
@@ -65,19 +65,73 @@ public class DataBufferThread extends Thread {
}
}
+ private void closeCurrentBufferFile(FileOutputStream fileOutputStream) {
+ try {
+ fileOutputStream.flush();
+ fileOutputStream.write(BufferDataPackagerGenerator.generateEOFPackage());
+ } catch (IOException e) {
+ logger.error("Failed to write msg.", e);
+ } finally {
+ try {
+ fileOutputStream.flush();
+ fileOutputStream.close();
+ } catch (IOException e) {
+ logger.error("Failed to flush and close file.", e);
+ }
+ }
+ }
+ private FileOutputStream acquiredNewBufferFileStream() {
+ checkBufferDirIsExists();
+ File outputFile = createNewBufferFile();
+ return generateFileOutputStream(outputFile);
+ }
+
+ private FileOutputStream generateFileOutputStream(File outputFile) {
+ FileOutputStream fileOutputStream = null;
+ try {
+ fileOutputStream = new FileOutputStream(outputFile);
+ } catch (FileNotFoundException e) {
+ logger.error("Failed to create File:{}", outputFile.getName(), e);
+ }
+ return fileOutputStream;
+ }
+
+ private File createNewBufferFile() {
+ File outputFile = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIR,
+ System.currentTimeMillis() + "-" + UUID.randomUUID().toString());
+ if (!outputFile.exists()) {
+ try {
+ outputFile.createNewFile();
+ ServerHealthCollector.getCurrentHeathReading(null)
+ .updateData(ServerHeathReading.INFO, "Create new Buffer file[" + outputFile.getName() + "]");
+ } catch (IOException e) {
+ logger.error("Failed to create File:{}", outputFile.getName(), e);
+ }
+ }
+ return outputFile;
+ }
+
+ private void checkBufferDirIsExists() {
+ File outPutDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIR);
+ if (!outPutDir.exists()) {
+ outPutDir.mkdirs();
+ }
+ }
public void saveTemporarily(byte[] s) {
int i = index.getAndIncrement();
while (data[i] != null) {
try {
- ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.WARNING, "DataBuffer index[" + i + "] data collision, service pausing. ");
+ ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.WARNING,
+ "DataBuffer index[" + i + "] data collision, service pausing. ");
Thread.sleep(DATA_CONFLICT_WAIT_TIME);
} catch (InterruptedException e) {
logger.error("Failure sleep.", e);
}
}
- ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer reveiving data.");
+ ServerHealthCollector.getCurrentHeathReading(null)
+ .updateData(ServerHeathReading.INFO, "DataBuffer reveiving data.");
data[i] = s;
}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java
index 98764a344..f40d4dd94 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java
@@ -1,13 +1,19 @@
package com.ai.cloud.skywalking.reciever.buffer;
import com.ai.cloud.skywalking.reciever.conf.Config;
+import org.apache.commons.io.comparator.NameFileComparator;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.io.File;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
+import static com.ai.cloud.skywalking.reciever.conf.Config.Persistence.MAX_APPEND_EOF_FLAGS_THREAD_NUMBER;
+
public class DataBufferThreadContainer {
private static Logger logger = LogManager.getLogger(DataBufferThreadContainer.class);
@@ -23,7 +29,38 @@ public class DataBufferThreadContainer {
return buffers.get(ThreadLocalRandom.current().nextInt(buffers.size()));
}
- public static void init() {
+ public static void init() throws InterruptedException {
+ logger.info("Add EOF flags to the unprocessed data file last time.");
+ File parentDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIR);
+ // 判断数据缓存文件是否存在，如果不存在，则创建
+ if (!parentDir.exists()) {
+ parentDir.mkdirs();
+ }
+ NameFileComparator sizeComparator = new NameFileComparator();
+ File[] dataFileList = sizeComparator.sort(parentDir.listFiles());
+ logger.info("Pending file number :" + dataFileList.length);
+ if (dataFileList.length > 0) {
+ int step = (int) Math.ceil(dataFileList.length * 1.0 / MAX_APPEND_EOF_FLAGS_THREAD_NUMBER);
+
+ int start = 0, end = 0;
+ CountDownLatch countDownLatch = new CountDownLatch(MAX_APPEND_EOF_FLAGS_THREAD_NUMBER);
+ while (true) {
+ if (end + step >= dataFileList.length) {
+ new AppendEOFFlagThread(Arrays.copyOfRange(dataFileList, start, dataFileList.length),
+ countDownLatch).start();
+ break;
+ }
+ end += step;
+ new AppendEOFFlagThread(Arrays.copyOfRange(dataFileList, start, end), countDownLatch).start();
+ start = end;
+ logger.debug("start:" + start + "\tend:" + end);
+ }
+
+ countDownLatch.await();
+ }
+ logger.info("Data buffer thread size {} begin to init ", Config.Server.
+ MAX_DEAL_DATA_THREAD_NUMBER);
+
for (int i = 0; i < Config.Server.MAX_DEAL_DATA_THREAD_NUMBER; i++) {
DataBufferThread dataBufferThread = new DataBufferThread(i);
dataBufferThread.start();
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java
index 234310c76..f2cb5d942 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java
@@ -17,6 +17,8 @@ public class Config {
// 数据缓存配置类
public static class Buffer {
+ // 数据缓存文件目录
+ public static String DATA_BUFFER_FILE_PARENT_DIR = "/tmp/skywalking/data/buffer";
//每个线程最大缓存数量
public static int PER_THREAD_MAX_BUFFER_NUMBER = 1024;
@@ -27,6 +29,33 @@ public class Config {
// 数据冲突时等待时间(单位:毫秒)
public static long DATA_CONFLICT_WAIT_TIME = 10L;
+ public static long BUFFER_FILE_MAX_LENGTH = 30 * 1024 * 1024;
+
+ }
+
+
+ public static class RegisterPersistence {
+ // 偏移量注册文件的目录
+ public static String REGISTER_FILE_PARENT_DIRECTORY = "../data/offset";
+
+ // 偏移量注册文件名
+ public static String REGISTER_FILE_NAME = "offset.A";
+
+ // 偏移量注册备份文件名
+ public static String REGISTER_BAK_FILE_NAME = "offset.B";
+
+ // 偏移量写入文件等待周期
+ public static long OFFSET_WRITTEN_FILE_WAIT_CYCLE = 5000L;
+ }
+
+
+ public static class Persistence {
+ // 切换文件，等待时间
+ public static long SWITCH_FILE_WAIT_TIME = 5000L;
+
+ // 追加EOF标志位的线程数量
+ public static int MAX_APPEND_EOF_FLAGS_THREAD_NUMBER = 1;
+
}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java
index 9bd89925f..958b5b095 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java
@@ -1,15 +1,15 @@
package com.ai.cloud.skywalking.reciever.handler;
+import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
import com.ai.cloud.skywalking.reciever.buffer.DataBufferThreadContainer;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.util.RedisConnector;
-import com.ai.cloud.skywalking.protocol.TransportPackager;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import redis.clients.jedis.Jedis;
import java.net.InetSocketAddress;
-import java.util.List;
+import java.util.Arrays;
public class CollectionServerDataHandler extends SimpleChannelInboundHandler {
@@ -18,11 +18,8 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler= 0) {
-
- List byteSerializeObjects = TransportPackager.unpack(msg);
-
- if (byteSerializeObjects.size() > 0) {
- cacheSerializeObjects(byteSerializeObjects);
+ if (validateCheckSum(msg)) {
+ DataBufferThreadContainer.getDataBufferThread().saveTemporarily(unpackCheckSum(msg));
} else {
// 处理错误包
dealFailedPackage(ctx);
@@ -30,12 +27,27 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler byteSerializeObjects) {
- for (byte[] byteSerializeObject : byteSerializeObjects) {
- DataBufferThreadContainer.getDataBufferThread().saveTemporarily(byteSerializeObject);
- }
+ private byte[] unpackCheckSum(byte[] msg) {
+ return Arrays.copyOfRange(msg, 4, msg.length);
}
+ private boolean validateCheckSum(byte[] dataPackage) {
+ byte[] checkSum = generateChecksum(dataPackage, 4);
+ byte[] originCheckSum = new byte[4];
+ System.arraycopy(dataPackage, 0, originCheckSum, 0, 4);
+ return Arrays.equals(checkSum, originCheckSum);
+ }
+
+ private static byte[] generateChecksum(byte[] data, int offset) {
+ int result = data[offset];
+ for (int i = offset + 1; i < data.length; i++) {
+ result ^= data[i];
+ }
+
+ return IntegerAssist.intToBytes(result);
+ }
+
+
private void dealFailedPackage(ChannelHandlerContext ctx) {
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().localAddress();
String key = ctx.name() + "-" + socketAddress.getHostName() + ":" + socketAddress.getPort();
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BufferDataPackagerGenerator.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BufferDataPackagerGenerator.java
new file mode 100644
index 000000000..a2ff30dd0
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BufferDataPackagerGenerator.java
@@ -0,0 +1,39 @@
+package com.ai.cloud.skywalking.reciever.model;
+
+import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
+
+public class BufferDataPackagerGenerator {
+
+ private static byte[] SPILT = new byte[] {127, 127, 127, 127};
+ private static byte[] EOF = null;
+
+ private BufferDataPackagerGenerator() {
+ //DO Nothing
+ }
+
+ public static byte[] generateEOFPackage() {
+ if (EOF != null) {
+ return EOF;
+ }
+
+ EOF = generatePackage("EOF".getBytes());
+ return EOF;
+ }
+
+ public static byte[] pack(byte[] msg) {
+ return generatePackage(msg);
+ }
+
+
+ private static byte[] generatePackage(byte[] msg) {
+ byte[] dataPackage = new byte[msg.length + 8];
+ // 前四位长度
+ System.arraycopy(IntegerAssist.intToBytes(msg.length), 0, dataPackage, 0, 4);
+ // 中间正文
+ System.arraycopy(msg, 0, dataPackage, 4, msg.length);
+ // 后四位特殊字符
+ System.arraycopy(SPILT, 0, dataPackage, msg.length + 4, 4);
+
+ return dataPackage;
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/BufferBalePlucker.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/BufferBalePlucker.java
new file mode 100644
index 000000000..e5a378da6
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/BufferBalePlucker.java
@@ -0,0 +1,227 @@
+package com.ai.cloud.skywalking.reciever.peresistent;
+
+import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
+import org.apache.commons.io.FileUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.Arrays;
+
+public class BufferBalePlucker {
+ private File bufferFile;
+ private FileInputStream bufferInputStream;
+ private int currentOffset;
+ private boolean hasNextBufferBale = false;
+ private static final byte[] SPILT_BALE_ARRAY = new byte[] {127, 127, 127, 127};
+ private static final byte[] EOF_BALE_ARRAY = "EOF".getBytes();
+ private int remainderLength = 0;
+ private byte[] remainderByte = null;
+ private Logger logger = LogManager.getLogger(BufferBalePlucker.class);
+ private PluckerStatus status = PluckerStatus.INITIAL;
+
+ public BufferBalePlucker(File bufferFile, int currentOffset) {
+ this.bufferFile = bufferFile;
+ this.currentOffset = currentOffset;
+
+ if (bufferFile.length() >= currentOffset) {
+ hasNextBufferBale = true;
+ }
+
+ try {
+ this.bufferInputStream = new FileInputStream(bufferFile);
+ bufferInputStream.skip(currentOffset);
+ } catch (IOException e) {
+ hasNextBufferBale = false;
+ }
+
+ }
+
+ public boolean hasNextBufferBale() {
+
+ if (status == PluckerStatus.SUSPEND) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ logger.error("Failed to sleep ", e);
+ }
+ }
+
+ if (status == PluckerStatus.INITIAL) {
+ status = PluckerStatus.RUNNING;
+ return hasNextBufferBale;
+ }
+
+ if (status == PluckerStatus.RUNNING) {
+ byte[] spiltArray = new byte[4];
+ try {
+ bufferInputStream.read(spiltArray);
+ if (!Arrays.equals(spiltArray, SPILT_BALE_ARRAY)) {
+ skipToNextBufferBale();
+ }
+ } catch (IOException e) {
+ return false;
+ }
+ currentOffset += 4;
+ }
+ MemoryRegister.instance().updateOffSet(bufferFile.getName(), currentOffset);
+ return hasNextBufferBale;
+ }
+
+ public byte[] pluck() throws IOException {
+ int packageLength = unpackBaleLength();
+ byte[] dataPackage = unpackDataContext(packageLength);
+
+ if (dataPackage == null || dataPackage.length == 0) {
+ // 文件没有完结,但是已经被处理完成了
+ return null;
+ }
+
+ if (checkDataPackageIsEOF(dataPackage)) {
+ hasNextBufferBale = false;
+ return null;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Pluck bale size : " + dataPackage.length);
+ }
+
+ return dataPackage;
+ }
+
+ private boolean checkDataPackageIsEOF(byte[] dataPackage) {
+ if (dataPackage.length == EOF_BALE_ARRAY.length) {
+ return Arrays.equals(dataPackage, EOF_BALE_ARRAY);
+ }
+ return false;
+ }
+
+ private byte[] unpackDataContext(int length) throws IOException {
+ if (currentOffset >= bufferFile.length()) {
+ status = PluckerStatus.SUSPEND;
+ return null;
+ }
+ byte[] dataContext = new byte[length];
+ bufferInputStream.read(dataContext);
+ currentOffset += length;
+ return dataContext;
+ }
+
+ private int unpackBaleLength() {
+ int length;
+
+ while (true) {
+ try {
+ length = calculateCurrentPackageLength();
+ if (length > 0 && length < 90000) {
+ break;
+ }
+ skipToNextBufferBale();
+ } catch (IOException e) {
+ skipToNextBufferBale();
+ }
+ }
+
+ return length;
+ }
+
+ private int calculateCurrentPackageLength() throws IOException {
+ byte[] lengthByte = new byte[4 - remainderLength];
+ bufferInputStream.read(lengthByte);
+ currentOffset += 4 - remainderLength;
+ lengthByte = spliceRemainderByteOfPreviousSkipIfNecessary(lengthByte);
+ return IntegerAssist.bytesToInt(lengthByte, 0);
+ }
+
+ private byte[] spliceRemainderByteOfPreviousSkipIfNecessary(byte[] lengthByte) {
+ if (remainderLength != 0) {
+ byte[] length = Arrays.copyOf(remainderByte, 4);
+ System.arraycopy(lengthByte, 0, length, remainderLength, lengthByte.length);
+ remainderLength = 0;
+ return length;
+ }
+ return lengthByte;
+ }
+
+ public void skipToNextBufferBale() {
+ byte[] previousDataByte = new byte[4];
+ byte[] currentDataByte = new byte[4];
+ byte[] compactDataByte = new byte[8];
+ while (true) {
+ try {
+ currentOffset += bufferInputStream.read(currentDataByte);
+ } catch (IOException e) {
+ hasNextBufferBale = false;
+ }
+
+ if (Arrays.equals(currentDataByte, SPILT_BALE_ARRAY)) {
+ remainderLength = 0;
+ break;
+ }
+
+ //
+ if (currentOffset + 8000 >= bufferFile.length()) {
+ status = PluckerStatus.SUSPEND;
+ break;
+ }
+
+ System.arraycopy(previousDataByte, 0, compactDataByte, 0, 4);
+ System.arraycopy(currentDataByte, 0, compactDataByte, 4, 4);
+
+ int index = bytesIndexOf(compactDataByte, SPILT_BALE_ARRAY, 0, 8);
+ if (index != -1) {
+ recodeRemainderByteAndLength(compactDataByte, index);
+ break;
+ }
+
+ previousDataByte = Arrays.copyOf(currentDataByte, 4);
+ }
+ }
+
+ private void recodeRemainderByteAndLength(byte[] compactDataByte, int index) {
+ remainderLength = 8 - 4 - index;
+ remainderByte = Arrays.copyOfRange(compactDataByte, index + 4, index + 4 + remainderLength);
+ }
+
+ private int bytesIndexOf(byte[] Source, byte[] Search, int fromIndex, int endIndex) {
+ boolean find = false;
+ int i;
+ for (i = fromIndex; i < endIndex - Search.length; i++) {
+ if (Source[i] == Search[0]) {
+ find = true;
+ for (int j = 0; j < Search.length; j++) {
+ if (Source[i + j] != Search[j]) {
+ find = false;
+ }
+ }
+ }
+ if (find) {
+ break;
+ }
+ }
+ return !find ? -1 : i;
+ }
+
+ public void close() throws IOException {
+ if (bufferInputStream != null) {
+ bufferInputStream.close();
+ }
+ logger.info("Data in file[{}] has been successfully processed", bufferFile.getName());
+ boolean deleteSuccess = false;
+ while (!deleteSuccess) {
+ deleteSuccess = FileUtils.deleteQuietly(new File(bufferFile.getParent(), bufferFile.getName()));
+ }
+ logger.info("Delete file[{}] {}", bufferFile.getName(), (deleteSuccess ? "success" : "failed"));
+ MemoryRegister.instance().removeEntry(bufferFile.getName());
+ bufferFile = null;
+ }
+
+ enum PluckerStatus {
+ INITIAL,
+ RUNNING,
+ SUSPEND
+ }
+
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/FileRegisterEntry.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/FileRegisterEntry.java
new file mode 100644
index 000000000..a284ddc2c
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/FileRegisterEntry.java
@@ -0,0 +1,78 @@
+package com.ai.cloud.skywalking.reciever.peresistent;
+
+public class FileRegisterEntry {
+ private String fileName;
+ private int offset;
+ private FileRegisterEntryStatus status;
+
+ public FileRegisterEntry() {
+ }
+
+ public FileRegisterEntry(String fileName) {
+ this.fileName = fileName;
+ }
+
+ public FileRegisterEntry(String fileName, int offset) {
+ this.fileName = fileName;
+ this.offset = offset;
+ }
+
+ public FileRegisterEntry(String fileName, int offset, FileRegisterEntryStatus status) {
+ this.fileName = fileName;
+ this.offset = offset;
+ this.status = status;
+ }
+
+ public String getFileName() {
+ return fileName;
+ }
+
+ public void setFileName(String fileName) {
+ this.fileName = fileName;
+ }
+
+ public int getOffset() {
+ return offset;
+ }
+
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
+
+ public FileRegisterEntryStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(FileRegisterEntryStatus status) {
+ this.status = status;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (!(o instanceof FileRegisterEntry))
+ return false;
+
+ FileRegisterEntry that = (FileRegisterEntry) o;
+
+ return !(getFileName() != null ? !getFileName().equals(that.getFileName()) : that.getFileName() != null);
+
+ }
+
+ @Override
+ public int hashCode() {
+ return getFileName() != null ? getFileName().hashCode() : 0;
+ }
+
+
+ @Override
+ public String toString() {
+ return fileName + '\t' + offset + "\t" + status;
+ }
+
+ public enum FileRegisterEntryStatus {
+ REGISTER,
+ UNREGISTER;
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/MemoryRegister.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/MemoryRegister.java
new file mode 100644
index 000000000..a5ee75cb5
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/MemoryRegister.java
@@ -0,0 +1,127 @@
+package com.ai.cloud.skywalking.reciever.peresistent;
+
+import com.ai.cloud.skywalking.reciever.conf.Config;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static com.ai.cloud.skywalking.reciever.conf.Config.RegisterPersistence.*;
+
+public class MemoryRegister {
+ private Logger logger = LogManager.getLogger(MemoryRegister.class);
+ private Map entries = new ConcurrentHashMap();
+ private File file;
+ private static MemoryRegister memoryRegister = new MemoryRegister();
+
+ public static MemoryRegister instance() {
+ return memoryRegister;
+ }
+
+ public void updateOffSet(String fileName, int offset) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Register entry[{}] offset[{}] into the memory register", fileName, offset);
+ }
+ if (entries.containsKey(fileName)) {
+ entries.get(fileName).setOffset(offset);
+ }
+ }
+
+ public void unRegister(String fileName) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Unregister[{}] from the memory register", fileName);
+ }
+ if (entries.containsKey(fileName)) {
+ entries.get(fileName).setStatus(FileRegisterEntry.FileRegisterEntryStatus.UNREGISTER);
+ }
+ }
+
+
+ public void removeEntry(String fileName) {
+ entries.remove(fileName);
+ }
+
+ public synchronized FileRegisterEntry doRegister(String fileName) {
+ logger.debug("Begin to register File[{}]", fileName);
+ FileRegisterEntry entry = null;
+ // 已经存在entries.
+ if (entries.containsKey(fileName)) {
+ logger.debug("FileRegisterEntry[{}] Status:[{}]", entries.get(fileName).getStatus());
+ // 已经被别的线程处理中
+ if (entries.get(fileName).getStatus() == FileRegisterEntry.FileRegisterEntryStatus.REGISTER) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Entry[{}] has been register", fileName);
+ }
+
+ } else {
+ // 没有被别的线程处理
+ entry = entries.get(fileName);
+ entry.setStatus(FileRegisterEntry.FileRegisterEntryStatus.REGISTER);
+ }
+ } else {
+ // 以前没有被注册过的
+ entry = new FileRegisterEntry(fileName, 0, FileRegisterEntry.FileRegisterEntryStatus.REGISTER);
+ entries.put(fileName, entry);
+ }
+
+ return entry;
+ }
+
+
+ public Collection getEntries() {
+ return entries.values();
+ }
+
+ public int getOffSet(String fileName) {
+ if (entries.containsKey(fileName)) {
+ return entries.get(fileName).getOffset();
+ }
+ return -1;
+ }
+
+ private void checkOffSetExists() {
+ file = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_FILE_NAME);
+
+ if (!file.exists()) {
+ try {
+ file.createNewFile();
+ } catch (IOException e) {
+ logger.error("Create offset filed failed", e);
+ }
+ }
+ }
+
+ private MemoryRegister() {
+ // 在处理数据之前需要初始化处理文件的处理状态
+ //去掉entries中无法与缓存数据文件匹配的文件
+ File parentDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIR);
+ //上次未处理的缓存数据文件，entries内的数据主要以缓存
+ List bufferFileNameList = Arrays.asList(parentDir.list());
+
+ try {
+ // 读取offset文件
+ OffsetFile mainOffsetFile = new OffsetFile(REGISTER_FILE_PARENT_DIRECTORY + File.separator +
+ REGISTER_FILE_NAME, bufferFileNameList);
+ OffsetFile backUpOffsetFile = new OffsetFile(REGISTER_FILE_PARENT_DIRECTORY + File.separator +
+ REGISTER_BAK_FILE_NAME, bufferFileNameList);
+
+ if (mainOffsetFile.compare(backUpOffsetFile)) {
+ entries.putAll(mainOffsetFile.getRegisterEntries());
+ } else {
+ entries.putAll(backUpOffsetFile.getRegisterEntries());
+ }
+ } catch (FileNotFoundException e) {
+ logger.error("The offset file does not exist.", e);
+ checkOffSetExists();
+ } catch (IOException e) {
+ logger.error("Read data from offset file failed.", e);
+ }
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/OffsetFile.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/OffsetFile.java
new file mode 100644
index 000000000..546aa0f8f
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/OffsetFile.java
@@ -0,0 +1,69 @@
+package com.ai.cloud.skywalking.reciever.peresistent;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class OffsetFile {
+ private boolean isComplete = true;
+ private Map registerEntries = new HashMap();
+ private long lastModifyTime;
+
+ public OffsetFile(String offsetFileName, List bufferFileNameList) throws IOException {
+ BufferedReader reader = new BufferedReader(new FileReader(offsetFileName));
+ String offsetData;
+ String lastModifyTimeStr = reader.readLine();
+ if (lastModifyTimeStr == null || lastModifyTimeStr.length() == 0) {
+ isComplete = false;
+ return;
+ }
+ lastModifyTime = Long.parseLong(lastModifyTimeStr);
+ while ((offsetData = reader.readLine()) != null && !"EOF".equals(offsetData)) {
+ String[] ss = offsetData.split("\t");
+ if (bufferFileNameList.contains(ss[0])) {
+ registerEntries.put(ss[0], new FileRegisterEntry(ss[0], Integer.valueOf(ss[1]),
+ FileRegisterEntry.FileRegisterEntryStatus.UNREGISTER));
+ }
+ }
+
+ if (!"EOF".equals(offsetData)) {
+ // 文件不完整
+ isComplete = false;
+ }
+ }
+
+ public boolean compare(OffsetFile offsetFile) {
+ if (isComplete && !offsetFile.isComplete) {
+ if (lastModifyTime > offsetFile.lastModifyTime) {
+ // 优先选择完整,并且时间是最新的
+ return true;
+ } else {
+ return false;
+ }
+ } else if (isComplete && offsetFile.isComplete) {
+ // 都是完整,则采用时间最新的
+ if (lastModifyTime > offsetFile.lastModifyTime) {
+ return true;
+ } else {
+ return false;
+ }
+ } else if (!isComplete && offsetFile.isComplete) {
+ //本身没有完成,但是别人完成了,采用别人的
+ return false;
+ } else {
+ // 自己没有完成,别人也没有完成,则比较时间
+ if (lastModifyTime > offsetFile.lastModifyTime) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+ }
+
+ public Map getRegisterEntries() {
+ return registerEntries;
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/PersistenceThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/PersistenceThread.java
new file mode 100644
index 000000000..7efbe2e60
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/PersistenceThread.java
@@ -0,0 +1,142 @@
+package com.ai.cloud.skywalking.reciever.peresistent;
+
+import com.ai.cloud.skywalking.protocol.SerializedFactory;
+import com.ai.cloud.skywalking.protocol.TransportPackager;
+import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
+import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
+import com.ai.cloud.skywalking.reciever.conf.Config;
+import com.ai.cloud.skywalking.reciever.processor.IProcessor;
+import com.ai.cloud.skywalking.reciever.processor.ProcessorFactory;
+import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
+import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
+import org.apache.commons.io.comparator.NameFileComparator;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PersistenceThread extends Thread {
+
+ private Logger logger = LogManager.getLogger(PersistenceThread.class);
+ private File bufferFile = null;
+
+ public PersistenceThread(int trdIndex) {
+ super("PersistentThread" + trdIndex);
+ }
+
+ @Override
+ public void run() {
+ while (true) {
+ bufferFile = chooseDealBufferFile();
+
+ if (bufferFile == null) {
+ try {
+ Thread.sleep(Config.Persistence.SWITCH_FILE_WAIT_TIME);
+ } catch (InterruptedException e) {
+ logger.error("Failure sleep.", e);
+ }
+ continue;
+ }
+
+ int offset = acquireOffset();
+
+ BufferBalePlucker bufferBalePlucker = new BufferBalePlucker(bufferFile, offset);
+ while (bufferBalePlucker.hasNextBufferBale()) {
+ try {
+ Map> spans = unSerializeSpans(bufferBalePlucker.pluck());
+ System.out.println(spans.size());
+ //handleSpans(spans);
+ } catch (ConvertFailedException e) {
+ bufferBalePlucker.skipToNextBufferBale();
+ } catch (IOException e) {
+ logger.error("The data file I/O exception.", e);
+ ServerHealthCollector.getCurrentHeathReading(null)
+ .updateData(ServerHeathReading.ERROR, e.getMessage());
+ }
+ }
+
+ try {
+ bufferBalePlucker.close();
+ } catch (IOException e) {
+ logger.error("The data file I/O exception.", e);
+ ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.ERROR, e.getMessage());
+ }
+
+ try {
+ Thread.sleep(Config.Persistence.SWITCH_FILE_WAIT_TIME);
+ } catch (InterruptedException e) {
+ logger.error("Failure sleep.", e);
+ }
+ }
+ }
+
+ private int acquireOffset() {
+ int offset;
+ offset = MemoryRegister.instance().getOffSet(bufferFile.getName());
+ if (offset == -1 || offset == 0) {
+ offset = 0;
+ }
+ return offset;
+ }
+
+ private void handleSpans(Map> spans) {
+ for (Map.Entry> entry : spans.entrySet()) {
+ IProcessor processor = ProcessorFactory.chooseProcessor(entry.getKey());
+ if (processor != null) {
+ processor.process(entry.getValue());
+ }
+ }
+ }
+
+ private Map> unSerializeSpans(byte[] byteData)
+ throws ConvertFailedException {
+ Map> spans = new HashMap>();
+ if (byteData == null || byteData.length == 0) {
+ return spans;
+ }
+ List serializeData = TransportPackager.unpackDataBody(byteData);
+ for (byte[] dataBytes : serializeData) {
+ AbstractDataSerializable abstractDataSerializable = SerializedFactory.unSerialize(dataBytes);
+ if (spans.get(abstractDataSerializable.getDataType()) == null) {
+ spans.put(abstractDataSerializable.getDataType(), new ArrayList());
+ }
+ spans.get(abstractDataSerializable.getDataType()).add(abstractDataSerializable);
+ }
+
+ return spans;
+ }
+
+ private File chooseDealBufferFile() {
+ File file1 = null;
+ File parentDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIR);
+ NameFileComparator sizeComparator = new NameFileComparator();
+ File[] dataFileList = sizeComparator.sort(parentDir.listFiles());
+
+ if (dataFileList == null) {
+ return null;
+ }
+
+ for (File file : dataFileList) {
+ if (file.getName().startsWith(".")) {
+ continue;
+ }
+ if (MemoryRegister.instance().doRegister(file.getName()) == null) {
+ if (logger.isDebugEnabled())
+ logger.debug("The file [{}] is being used by another thread ", file);
+ continue;
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("Begin to deal data file [{}]", file.getName());
+ }
+ file1 = file;
+ break;
+ }
+
+ return file1;
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/PersistenceThreadLauncher.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/PersistenceThreadLauncher.java
new file mode 100644
index 000000000..8f3d2e526
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/PersistenceThreadLauncher.java
@@ -0,0 +1,12 @@
+package com.ai.cloud.skywalking.reciever.peresistent;
+
+import com.ai.cloud.skywalking.reciever.conf.Config;
+
+public class PersistenceThreadLauncher {
+ public static void doLaunch() {
+ new RegisterPersistenceThread().start();
+ for (int i = 0; i < Config.Server.MAX_DEAL_DATA_THREAD_NUMBER; i++) {
+ new PersistenceThread(i).start();
+ }
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/RegisterPersistenceThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/RegisterPersistenceThread.java
new file mode 100644
index 000000000..ab2e7fc87
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/RegisterPersistenceThread.java
@@ -0,0 +1,98 @@
+package com.ai.cloud.skywalking.reciever.peresistent;
+
+import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
+import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Collection;
+
+import static com.ai.cloud.skywalking.reciever.conf.Config.RegisterPersistence.*;
+
+public class RegisterPersistenceThread extends Thread {
+ private Logger logger = LogManager.getLogger(RegisterPersistenceThread.class);
+ private File offsetFile;
+ private File bakOffsetFile;
+ private BufferedWriter offsetBufferWriter;
+ private boolean isSwitch = true;
+
+ public RegisterPersistenceThread() {
+ super("RegisterPersistenceThread");
+ File offsetParentDir = new File(REGISTER_FILE_PARENT_DIRECTORY);
+ if (!offsetParentDir.exists()) {
+ offsetParentDir.mkdirs();
+ }
+
+ offsetFile = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_FILE_NAME);
+ bakOffsetFile = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_BAK_FILE_NAME);
+ }
+
+ @Override
+ public void run() {
+
+ while (true) {
+ try {
+ Thread.sleep(OFFSET_WRITTEN_FILE_WAIT_CYCLE);
+ } catch (InterruptedException e) {
+ logger.error("Sleep failure", e);
+ }
+ chooseOffsetFile();
+ appendingTimestamp();
+ wirteFileRegisterEntries();
+ closeAndReleaseResource();
+ ServerHealthCollector.getCurrentHeathReading(null)
+ .updateData(ServerHeathReading.INFO, "flush memory register to file.");
+ }
+ }
+
+ private void chooseOffsetFile() {
+ try {
+ if (isSwitch) {
+ offsetBufferWriter = new BufferedWriter(new FileWriter(offsetFile));
+ } else {
+ offsetBufferWriter = new BufferedWriter(new FileWriter(bakOffsetFile));
+ }
+ isSwitch = !isSwitch;
+ } catch (IOException e) {
+ logger.error("Write The offset file anomalies.");
+ }
+ }
+
+ private void appendingTimestamp() {
+ try {
+ offsetBufferWriter.write(String.valueOf(System.currentTimeMillis()) + "\n");
+ } catch (IOException e) {
+ logger.error("Write The offset file anomalies.");
+ }
+ }
+
+ private void wirteFileRegisterEntries() {
+ Collection fileRegisterEntries = MemoryRegister.instance().getEntries();
+ for (FileRegisterEntry fileRegisterEntry : fileRegisterEntries) {
+ try {
+ offsetBufferWriter.write(fileRegisterEntry.toString() + "\n");
+ } catch (IOException e) {
+ logger.error("Write file register entry to offset file failure", e);
+ }
+ }
+ }
+
+ private void closeAndReleaseResource() {
+ try {
+ offsetBufferWriter.write("EOF\n");
+ offsetBufferWriter.flush();
+ } catch (IOException e) {
+ logger.error("Flush offset file failure", e);
+ } finally {
+ try {
+ offsetBufferWriter.close();
+ } catch (IOException e) {
+ logger.error("close offset file failure", e);
+ }
+ }
+ }
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/AckSpanProcessor.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/AckSpanProcessor.java
index 14c78c715..68aa284db 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/AckSpanProcessor.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/AckSpanProcessor.java
@@ -14,6 +14,7 @@ import java.util.List;
import static com.ai.cloud.skywalking.reciever.util.SpanUtil.getTSBySpanTraceId;
+@DefaultProcessor
public class AckSpanProcessor extends AbstractSpanProcessor {
@Override
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/DefaultProcessor.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/DefaultProcessor.java
new file mode 100644
index 000000000..e68e5d933
--- /dev/null
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/DefaultProcessor.java
@@ -0,0 +1,12 @@
+package com.ai.cloud.skywalking.reciever.processor;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface DefaultProcessor {
+ boolean defaultProcessor() default true;
+}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/ProcessorFactory.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/ProcessorFactory.java
index e1694a12c..d48eff01b 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/ProcessorFactory.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/ProcessorFactory.java
@@ -8,15 +8,19 @@ import java.util.Map;
import java.util.ServiceLoader;
public class ProcessorFactory {
- private static Logger logger = LogManager.getLogger(ProcessorFactory.class);
+ private static Logger logger = LogManager.getLogger(ProcessorFactory.class);
private static Map type_processor_mapping = new HashMap();
static {
ServiceLoader processors = ServiceLoader.load(IProcessor.class);
-
for (IProcessor processor : processors) {
- logger.info("Init protocol type and processor mapping : {} --> {}.", processor.getProtocolType(), processor.getClass().getName());
- type_processor_mapping.put(processor.getProtocolType(), processor);
+ DefaultProcessor defaultProcessor = processor.getClass().getAnnotation(DefaultProcessor.class);
+ IProcessor processor1 = type_processor_mapping.get(processor.getProtocolType());
+ if (processor1 == null || (defaultProcessor != null && !defaultProcessor.defaultProcessor())) {
+ logger.info("Init protocol type and processor mapping : {} --> {}.", processor.getProtocolType(),
+ processor.getClass().getName());
+ type_processor_mapping.put(processor.getProtocolType(), processor);
+ }
}
}
diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/RequestSpanProcessor.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/RequestSpanProcessor.java
index 715186a37..e46aa9509 100644
--- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/RequestSpanProcessor.java
+++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/RequestSpanProcessor.java
@@ -14,6 +14,7 @@ import java.util.List;
import static com.ai.cloud.skywalking.reciever.util.SpanUtil.getTSBySpanTraceId;
+@DefaultProcessor
public class RequestSpanProcessor extends AbstractSpanProcessor {
@@ -35,7 +36,8 @@ public class RequestSpanProcessor extends AbstractSpanProcessor {
} else {
columnName = requestSpan.getParentLevel() + "." + requestSpan.getLevelId();
}
- put.addColumn(Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME), Bytes.toBytes(columnName), requestSpan.getData());
+ put.addColumn(Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME), Bytes.toBytes(columnName),
+ requestSpan.getData());
puts.add(put);
}
diff --git a/skywalking-server/src/main/resources/config.properties b/skywalking-server/src/main/resources/config.properties
index a525da474..76995a6e2 100644
--- a/skywalking-server/src/main/resources/config.properties
+++ b/skywalking-server/src/main/resources/config.properties
@@ -11,16 +11,26 @@ buffer.max_wait_time=5000
#数据冲突时等待时间(单位:毫秒)
buffer.data_conflict_wait_time=10
#数据缓存文件目录
-buffer.data_buffer_file_parent_directory=D:/test-data/data/buffer
+buffer.data_buffer_file_parent_dir=/tmp/skywalking/data/buffer
#缓存数据文件最大长度(单位:byte)
-buffer.data_file_max_length=104857600
-#每次Flush的缓存数据的个数
-buffer.flush_number_of_cache=30
+buffer.buffer_file_max_length=104857600
#每次缓存数据写入失败，最大尝试时间
buffer.write_data_failure_retry_interval = 10000
-#数据包的最大限制
-datapackage.max_data_package=1048576
+#切换数据文件，等待时间(单位:毫秒)
+persistence.switch_file_wait_time=5000
+#追加EOF标志位的线程数量
+persistence.max_append_eof_flags_thread_number=1
+
+#偏移量注册文件的目录
+registerpersistence.register_file_parent_directory=/tmp/skywalking/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-new
@@ -50,4 +60,4 @@ redis.edis_max_total=20
#告警检查器：异常告警检查
alarm.checker.turn_on_exception_checker=true
#告警检查器：执行时间超时告警检查
-alarm.checker.turn_on_execute_time_checker=true
\ No newline at end of file
+alarm.checker.turn_on_execute_time_checker=true
diff --git a/skywalking-server/src/test/java/test/com/ai/cloud/skywalking/reciever/peresistent/BufferBalePluckerTest.java b/skywalking-server/src/test/java/test/com/ai/cloud/skywalking/reciever/peresistent/BufferBalePluckerTest.java
new file mode 100644
index 000000000..0f644efd4
--- /dev/null
+++ b/skywalking-server/src/test/java/test/com/ai/cloud/skywalking/reciever/peresistent/BufferBalePluckerTest.java
@@ -0,0 +1,74 @@
+package test.com.ai.cloud.skywalking.reciever.peresistent;
+
+import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
+import com.ai.cloud.skywalking.reciever.peresistent.BufferBalePlucker;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertNull;
+
+public class BufferBalePluckerTest {
+
+ private File file;
+
+ @Before
+ public void initData() throws IOException {
+ file = new File("/tmp", "test.file");
+ if (!file.exists()) {
+ file.createNewFile();
+ }
+
+ FileOutputStream fileOutputStream = new FileOutputStream(file);
+ byte[] bytes = "HelloWorld".getBytes();
+ byte[] packageByte = Arrays.copyOf(IntegerAssist.intToBytes(bytes.length), bytes.length + 4 + 4);
+ System.arraycopy(bytes, 0, packageByte, 4, bytes.length);
+ System.arraycopy(new byte[] {127, 127, 127, 127}, 0, packageByte, bytes.length + 4, 4);
+
+ fileOutputStream.write(packageByte);
+ fileOutputStream.write(generatePackage("EOF".getBytes()));
+ fileOutputStream.close();
+ }
+
+ private static byte[] generatePackage(byte[] msg) {
+ byte[] dataPackage = new byte[msg.length + 8];
+ // 前四位长度
+ System.arraycopy(IntegerAssist.intToBytes(msg.length), 0, dataPackage, 0, 4);
+ // 中间正文
+ System.arraycopy(msg, 0, dataPackage, 4, msg.length);
+ // 后四位特殊字符
+ System.arraycopy(new byte[] {127, 127, 127, 127}, 0, dataPackage, msg.length + 4, 4);
+
+ return dataPackage;
+ }
+
+ @Test
+ public void pluck() throws Exception {
+ BufferBalePlucker plucker = new BufferBalePlucker(file, 0);
+ while (plucker.hasNextBufferBale()) {
+ byte[] data = plucker.pluck();
+ if (data == null)
+ continue;
+ assertArrayEquals("HelloWorld".getBytes(), data);
+ }
+ }
+
+ @Test
+ public void skipToNextBufferBale() throws Exception {
+ BufferBalePlucker plucker = new BufferBalePlucker(file, 3);
+ plucker.skipToNextBufferBale();
+ assertNull(plucker.pluck());
+ }
+
+ @After
+ public void clearData() {
+ file.deleteOnExit();
+ }
+
+}