增加将接受到的数据缓存到缓存文件中
This commit is contained in:
ascrutae
parent
f0058ab74c
commit
9c23c745c7
|
|
@ -86,7 +86,7 @@ public class CurrentThreadSpanStack {
|
|||
|
||||
public void invalidatePresentSpans() {
|
||||
for (SpanNode spanNode : spans) {
|
||||
spanNode.getData().setInvalidate(true);
|
||||
spanNode.getData().setValidate(true);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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())) {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1206,6 +1206,20 @@ public final class TraceProtocol {
|
|||
*/
|
||||
com.google.protobuf.ByteString
|
||||
getBussinessKeyBytes();
|
||||
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
boolean hasAgentId();
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
java.lang.String getAgentId();
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
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_;
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
public boolean hasAgentId() {
|
||||
return ((bitField0_ & 0x00000800) == 0x00000800);
|
||||
}
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
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;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
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_ = "";
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
public boolean hasAgentId() {
|
||||
return ((bitField0_ & 0x00000800) == 0x00000800);
|
||||
}
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
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;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
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;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
public Builder setAgentId(
|
||||
java.lang.String value) {
|
||||
if (value == null) {
|
||||
throw new NullPointerException();
|
||||
}
|
||||
bitField0_ |= 0x00000800;
|
||||
agentId_ = value;
|
||||
onChanged();
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
public Builder clearAgentId() {
|
||||
bitField0_ = (bitField0_ & ~0x00000800);
|
||||
agentId_ = getDefaultInstance().getAgentId();
|
||||
onChanged();
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* <code>required string agentId = 12;</code>
|
||||
*/
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -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 {
|
|||
* 当前调用链的本机描述<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/>
|
||||
* 1：异常<br/>
|
||||
* 异常判断原则：代码产生exception，并且此exception不在忽略列表中
|
||||
*/
|
||||
private byte statusCode = 0;
|
||||
private byte statusCode = 0;
|
||||
/**
|
||||
* 节点调用的错误堆栈<br/>
|
||||
* 堆栈以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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<Integer, AbstractDataSerializable> serializableMap = new HashMap<Integer, AbstractDataSerializable>();
|
||||
public static Map<Integer, AbstractDataSerializable> serializableMap =
|
||||
new HashMap<Integer, AbstractDataSerializable>();
|
||||
|
||||
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ public class Span {
|
|||
* 节点的状态<br/>
|
||||
* 不参与序列化
|
||||
*/
|
||||
protected boolean isInvalidate = false;
|
||||
protected boolean isValidate = true;
|
||||
|
||||
/**
|
||||
* 节点调用过程中的业务字段<br/>
|
||||
|
|
@ -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<String> 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;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,44 +16,30 @@ public class TransportPackager {
|
|||
return dataPackage;
|
||||
}
|
||||
|
||||
public static List<byte[]> unpack(byte[] dataPackage) {
|
||||
if (validateCheckSum(dataPackage)) {
|
||||
return unpackDataBody(unpackCheckSum(dataPackage));
|
||||
} else {
|
||||
return new ArrayList<byte[]>();
|
||||
}
|
||||
}
|
||||
|
||||
private static byte[] unpackCheckSum(byte[] dataPackage) {
|
||||
return Arrays.copyOfRange(dataPackage, 4, dataPackage.length);
|
||||
}
|
||||
|
||||
private static List<byte[]> unpackDataBody(byte[] dataPackage) {
|
||||
List<byte[]> serializeData = new ArrayList<byte[]>();
|
||||
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<byte[]> unpackDataBody(byte[] dataPackage) {
|
||||
List<byte[]> serializeData = null;
|
||||
try {
|
||||
serializeData = new ArrayList<byte[]>();
|
||||
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);
|
||||
}
|
||||
|
||||
/**
|
||||
* 生成校验和参数
|
||||
*
|
||||
|
|
|
|||
|
|
@ -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)) {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,4 @@
|
|||
package com.ai.cloud.skywalking.protocol.exception;
|
||||
|
||||
public class ConvertFailedException extends Exception {
|
||||
}
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,11 +18,6 @@
|
|||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.ai.cloud</groupId>
|
||||
<artifactId>skywalking-api</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.ai.cloud</groupId>
|
||||
<artifactId>skywalking-api</artifactId>
|
||||
|
|
|
|||
|
|
@ -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<io.netty.channel.socket.SocketChannel>() {
|
||||
@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.....");
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Integer, List<AbstractDataSerializable>> serializeObjects;
|
||||
FileOutputStream fileOutputStream = null;
|
||||
int length = 0;
|
||||
while (true) {
|
||||
serializeObjects = new HashMap<Integer, List<AbstractDataSerializable>>();
|
||||
for (int i = 0; i < data.length; i++) {
|
||||
if (data[i] == null) {
|
||||
continue;
|
||||
}
|
||||
|
||||
AbstractDataSerializable serializeData = SerializedFactory.unSerialize(data[i]);
|
||||
List<AbstractDataSerializable> hasBeenSerializedObjects = serializeObjects.get(serializeData.getDataType());
|
||||
if (hasBeenSerializedObjects == null) {
|
||||
serializeObjects.put(serializeData.getDataType(), new ArrayList<AbstractDataSerializable>());
|
||||
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<Integer, List<AbstractDataSerializable>> 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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<byte[]> {
|
||||
|
||||
|
|
@ -18,11 +18,8 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler<byt
|
|||
Thread.currentThread().setName("ServerReceiver");
|
||||
// 当接受到这条消息的是空，则忽略
|
||||
if (msg != null && msg.length >= 0) {
|
||||
|
||||
List<byte[]> 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<byt
|
|||
}
|
||||
}
|
||||
|
||||
private void cacheSerializeObjects(List<byte[]> 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();
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String, FileRegisterEntry> entries = new ConcurrentHashMap<String, FileRegisterEntry>();
|
||||
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<FileRegisterEntry> 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<String> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String, FileRegisterEntry> registerEntries = new HashMap<String, FileRegisterEntry>();
|
||||
private long lastModifyTime;
|
||||
|
||||
public OffsetFile(String offsetFileName, List<String> 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<String, FileRegisterEntry> getRegisterEntries() {
|
||||
return registerEntries;
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Integer, List<AbstractDataSerializable>> 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<Integer, List<AbstractDataSerializable>> spans) {
|
||||
for (Map.Entry<Integer, List<AbstractDataSerializable>> entry : spans.entrySet()) {
|
||||
IProcessor processor = ProcessorFactory.chooseProcessor(entry.getKey());
|
||||
if (processor != null) {
|
||||
processor.process(entry.getValue());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Map<Integer, List<AbstractDataSerializable>> unSerializeSpans(byte[] byteData)
|
||||
throws ConvertFailedException {
|
||||
Map<Integer, List<AbstractDataSerializable>> spans = new HashMap<Integer, List<AbstractDataSerializable>>();
|
||||
if (byteData == null || byteData.length == 0) {
|
||||
return spans;
|
||||
}
|
||||
List<byte[]> 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<AbstractDataSerializable>());
|
||||
}
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<FileRegisterEntry> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -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<Integer, IProcessor> type_processor_mapping = new HashMap<Integer, IProcessor>();
|
||||
|
||||
static {
|
||||
ServiceLoader<IProcessor> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
alarm.checker.turn_on_execute_time_checker=true
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue