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

调整服务端代码以及逻辑并解决API工程编译问题（待测试）

This commit is contained in:
ascrutae 2016-07-06 14:58:14 +08:00
parent 481f1a5881
commit beb6276adf
53 changed files with 3566 additions and 2979 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/ContextBuffer.java
View File

@ -2,7 +2,6 @@ package com.ai.cloud.skywalking.buffer;
import com.ai.cloud.skywalking.logging.LogManager;
import com.ai.cloud.skywalking.logging.Logger;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import java.util.concurrent.ThreadLocalRandom;

2
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/CurrentThreadSpanStack.java
View File

@ -86,7 +86,7 @@ public class CurrentThreadSpanStack {
public void invalidatePresentSpans() {
for (SpanNode spanNode : spans) {
spanNode.getData().setIsInvalidate(true);
spanNode.getData().setInvalidate(true);
}
}
}

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

@ -1,6 +1,5 @@
package com.ai.cloud.skywalking.invoke.monitor;
import com.ai.cloud.skywalking.buffer.BufferGroup;
import com.ai.cloud.skywalking.buffer.ContextBuffer;
import com.ai.cloud.skywalking.conf.AuthDesc;
import com.ai.cloud.skywalking.conf.Config;
@ -8,6 +7,7 @@ import com.ai.cloud.skywalking.context.CurrentThreadSpanStack;
import com.ai.cloud.skywalking.logging.LogManager;
import com.ai.cloud.skywalking.logging.Logger;
import com.ai.cloud.skywalking.model.ContextData;
import com.ai.cloud.skywalking.model.Identification;
import com.ai.cloud.skywalking.protocol.AckSpan;
import com.ai.cloud.skywalking.protocol.RequestSpan;
import com.ai.cloud.skywalking.protocol.Span;
@ -19,23 +19,24 @@ import static com.ai.cloud.skywalking.conf.Config.BuriedPoint.EXCLUSIVE_EXCEPTIO
public abstract class BaseInvokeMonitor {
private static Logger logger = LogManager
.getLogger(BaseInvokeMonitor.class);
private static Logger logger = LogManager.getLogger(BaseInvokeMonitor.class);
private static String EXCEPTION_SPLIT = ",";
private static Set<String> exclusiveExceptionSet = null;
protected ContextData beforeInvoke(Span spanData) {
protected ContextData beforeInvoke(Span spanData, Identification id) {
if (Config.BuriedPoint.PRINTF) {
logger.debug("TraceId:" + spanData.getTraceId()
+ "\tviewpointId:" + spanData.getViewPointId()
+ "\tParentLevelId:" + spanData.getParentLevel()
+ "\tLevelId:" + spanData.getLevelId());
logger.debug("TraceId:" + spanData.getTraceId() + "\tviewpointId:" + id.getViewPoint() + "\tParentLevelId:" + spanData.getParentLevel() + "\tLevelId:" + spanData
.getLevelId());
}
// 根据SpanData生成RequestSpan，并保存
ContextBuffer.save(new RequestSpan(spanData));
ContextBuffer.save(RequestSpan.RequestSpanBuilder.
newBuilder(spanData).callType(id.getCallType())
.viewPoint(id.getViewPoint())
.spanTypeDesc(id.getSpanTypeDesc())
.build());
// 将新创建的Context存放到ThreadLocal栈中。
CurrentThreadSpanStack.push(spanData);
@ -54,16 +55,10 @@ public abstract class BaseInvokeMonitor {
return;
}
// 加上花费时间
spanData.setCost(System.currentTimeMillis()
- spanData.getStartDate());
if (Config.BuriedPoint.PRINTF) {
logger.debug("TraceId:" + spanData.getTraceId()
+ "\tviewpointId:" + spanData.getViewPointId()
+ "\tParentLevelId:" + spanData.getParentLevel()
+ "\tLevelId:" + spanData.getLevelId()
+ "\tbusinessKey:" + spanData.getParameters());
logger.debug(
"TraceId-ACK:" + spanData.getTraceId() + "\tParentLevelId:" + spanData.getParentLevel() + "\tLevelId:" + spanData
.getLevelId() + "\tbusinessKey:" + spanData.getParameters());
}
// 生成并保存到缓存
ContextBuffer.save(new AckSpan(spanData));
@ -77,8 +72,7 @@ public abstract class BaseInvokeMonitor {
if (exclusiveExceptionSet == null) {
Set<String> exclusiveExceptions = new HashSet<String>();
String[] exceptions = EXCLUSIVE_EXCEPTIONS
.split(EXCEPTION_SPLIT);
String[] exceptions = EXCLUSIVE_EXCEPTIONS.split(EXCEPTION_SPLIT);
for (String exception : exceptions) {
exclusiveExceptions.add(exception);
}
@ -86,8 +80,7 @@ public abstract class BaseInvokeMonitor {
}
Span span = CurrentThreadSpanStack.peek();
span.handleException(th, exclusiveExceptionSet,
Config.BuriedPoint.MAX_EXCEPTION_STACK_LENGTH);
span.handleException(th, exclusiveExceptionSet, Config.BuriedPoint.MAX_EXCEPTION_STACK_LENGTH);
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}

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

@ -21,7 +21,7 @@ public class LocalMethodInvokeMonitor extends BaseInvokeMonitor {
Span spanData = ContextGenerator.generateSpanFromThreadLocal(id);
return super.beforeInvoke(spanData);
return super.beforeInvoke(spanData,id);
} catch (Throwable t) {
logger.error(t.getMessage(), t);
return new EmptyContextData();

27
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ContextGenerator.java
View File

@ -1,6 +1,5 @@
package com.ai.cloud.skywalking.util;
import com.ai.cloud.skywalking.conf.Config;
import com.ai.cloud.skywalking.context.CurrentThreadSpanStack;
import com.ai.cloud.skywalking.model.ContextData;
import com.ai.cloud.skywalking.model.Identification;
@ -16,7 +15,7 @@ public final class ContextGenerator {
*/
public static Span generateSpanFromThreadLocal(Identification id) {
Span spanData = getSpanFromThreadLocal();
initNewSpanData(spanData, id);
spanData.setStartDate(System.currentTimeMillis());
return spanData;
}
@ -32,28 +31,17 @@ public final class ContextGenerator {
// 校验传入的参数是否为空，如果为空，则新创建一个
if (context == null || StringUtil.isEmpty(context.getTraceId())) {
// 不存在，新创建一个Context
spanData = new Span(TraceIdGenerator.generate(), Config.SkyWalking.APPLICATION_CODE, Config.SkyWalking.USER_ID);
spanData = new Span(TraceIdGenerator.generate());
} else {
// 如果不为空，则将当前的Context存放到上下文
spanData = new Span(context.getTraceId(), context.getParentLevel(), context.getLevelId(), Config.SkyWalking.APPLICATION_CODE, Config.SkyWalking.USER_ID);
spanData = new Span(context.getTraceId(), context.getParentLevel(), context.getLevelId());
}
initNewSpanData(spanData, id);
spanData.setStartDate(System.currentTimeMillis());
return spanData;
}
private static void initNewSpanData(Span spanData, Identification id) {
spanData.setSpanTypeDesc(id.getSpanTypeDesc());
spanData.setViewPointId(id.getViewPoint());
spanData.setBusinessKey(id.getBusinessKey());
//FIX Add Call Type field
spanData.setCallType(id.getCallType());
// 设置基本信息
spanData.setStartDate(System.currentTimeMillis());
spanData.setProcessNo(BuriedPointMachineUtil.getProcessNo());
spanData.setAddress(BuriedPointMachineUtil.getHostDesc());
}
private static Span getSpanFromThreadLocal() {
Span span;
// 1.获取Context，从ThreadLocal栈中获取中
@ -61,14 +49,13 @@ public final class ContextGenerator {
// 2 校验Context，Context是否存在
if (parentSpan == null) {
// 不存在，新创建一个Context
span = new Span(TraceIdGenerator.generate(), Config.SkyWalking.APPLICATION_CODE, Config.SkyWalking.USER_ID);
span = new Span(TraceIdGenerator.generate());
} else {
// 根据ParentContextData的TraceId和RPCID
// LevelId是由SpanNode类的nextSubSpanLevelId字段进行初始化的.
// 所以在这里不需要初始化
span = new Span(parentSpan.getTraceId(), Config.SkyWalking.APPLICATION_CODE, Config.SkyWalking.USER_ID);
span = new Span(parentSpan.getTraceId());
// check parent span is RPC span
// if true, current span is invalidate and current span also belong to RPC span

1
skywalking-collector/skywalking-api/src/test/java/test/ai/cloud/assertspandata/SDKGeneratedDataTest.java
View File

@ -2,7 +2,6 @@ package test.ai.cloud.assertspandata;
import com.ai.cloud.skywalking.buffer.ContextBuffer;
import com.ai.cloud.skywalking.conf.Config;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.skywalking.testframework.api.TraceTreeAssert;
import org.junit.Test;

3
skywalking-collector/skywalking-protocol/pom.xml
View File

@ -87,11 +87,12 @@
<mkdir dir="src/main/gen-java" />
<path id="proto.path">
<fileset dir="src/main/proto">
<include name="**/*.proto" />
<include name="*.proto" />
</fileset>
</path>
<pathconvert pathsep=" " property="proto.files"
refid="proto.path" />
<exec executable="protoc">
<arg value="--java_out=src/main/gen-java" />
<arg value="-I${project.basedir}/src/main/proto" />

1100
skywalking-collector/skywalking-protocol/src/main/gen-java/com/ai/cloud/skywalking/protocol/proto/AckSpanProtobuf.java
View File

File diff suppressed because it is too large Load Diff

2848
skywalking-collector/skywalking-protocol/src/main/gen-java/com/ai/cloud/skywalking/protocol/proto/TraceProtocol.java Normal file
View File

File diff suppressed because it is too large Load Diff

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

@ -1,6 +1,8 @@
package com.ai.cloud.skywalking.protocol;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.proto.TraceProtocol;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.HashMap;
import java.util.Map;
@ -55,6 +57,10 @@ public class AckSpan extends AbstractDataSerializable {
*/
private Map<String, String> paramters = new HashMap<String, String>();
public AckSpan() {
}
public String getTraceId() {
return traceId;
}
@ -121,8 +127,43 @@ public class AckSpan extends AbstractDataSerializable {
return new byte[0];
}
@Override
public AbstractDataSerializable convertData(byte[] data) {
AckSpan ackSpan = new AckSpan();
try {
TraceProtocol.AckSpan ackSpanProtocol = TraceProtocol.AckSpan.parseFrom(data);
ackSpan.setTraceId(ackSpanProtocol.getTraceId());
ackSpan.setParentLevel(ackSpanProtocol.getParentLevel());
ackSpan.setLevelId(ackSpanProtocol.getLevelId());
ackSpan.setCost(ackSpanProtocol.getCost());
ackSpan.setExceptionStack(ackSpanProtocol.getExceptionStack());
ackSpan.setStatusCode((byte) ackSpanProtocol.getStatusCode());
} catch (InvalidProtocolBufferException e) {
return null;
}
return ackSpan;
}
@Override
public boolean isNull() {
return false;
}
public static class AckSpanBuilder {
private AckSpan ackSpan;
private AckSpanBuilder(Span span) {
ackSpan = new AckSpan(span);
}
public static AckSpanBuilder newBuilder(Span span) {
return new AckSpanBuilder(span);
}
public AckSpanBuilder countCost(long startTime) {
ackSpan.cost = System.currentTimeMillis() - startTime;
return this;
}
}
}

109
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSpan.java
View File

@ -3,6 +3,8 @@ 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.proto.TraceProtocol;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.HashMap;
import java.util.Map;
@ -24,7 +26,7 @@ public class RequestSpan extends AbstractDataSerializable {
* 当前调用链的本机描述<br/>
* 如当前序号为：0.1.0时，levelId=0
*/
private int levelId = 0;
private int levelId = 0;
/**
* 调用链中单个节点的入口描述<br/>
* 如：java方法名，调用的RPC地址等等
@ -33,12 +35,7 @@ public class RequestSpan extends AbstractDataSerializable {
/**
* 节点调用开始时间
*/
private long startDate = System.currentTimeMillis();
/**
* 节点调用的发生机器描述<br/>
* 包含机器名 + IP地址
*/
private String address = "";
private long startDate = System.currentTimeMillis();
/**
* 节点类型描述<br/>
@ -60,10 +57,6 @@ public class RequestSpan extends AbstractDataSerializable {
*/
private SpanType spanType = SpanType.LOCAL;
/**
* 节点调用的所在进程号
*/
private String processNo = "";
/**
* 节点调用所在的系统逻辑名称<br/>
* 由授权文件指定
@ -85,17 +78,16 @@ public class RequestSpan extends AbstractDataSerializable {
this.traceId = spanData.getTraceId();
this.parentLevel = spanData.getParentLevel();
this.levelId = spanData.getLevelId();
this.address = spanData.getAddress();
this.applicationId = spanData.getApplicationId();
this.callType = spanData.getCallType();
this.spanType = spanData.getSpanType();
this.spanTypeDesc = spanData.getSpanTypeDesc();
this.userId = spanData.getUserId();
if (isEntrySpan(spanData)) {
this.paramters.putAll(spanData.getParamters());
this.paramters.putAll(spanData.getParameters());
}
}
public RequestSpan() {
}
private boolean isEntrySpan(Span spanData) {
return "0".equals(spanData.getParentLevel() + spanData.getLevelId());
}
@ -140,14 +132,6 @@ public class RequestSpan extends AbstractDataSerializable {
this.startDate = startDate;
}
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
public String getSpanTypeDesc() {
return spanTypeDesc;
}
@ -172,14 +156,6 @@ public class RequestSpan extends AbstractDataSerializable {
this.spanType = spanType;
}
public String getProcessNo() {
return processNo;
}
public void setProcessNo(String processNo) {
this.processNo = processNo;
}
public String getApplicationId() {
return applicationId;
}
@ -211,11 +187,76 @@ public class RequestSpan extends AbstractDataSerializable {
@Override
public byte[] getData() {
return new byte[0];
return TraceProtocol.RequestSpan.newBuilder().setTraceId(traceId).setParentLevel(parentLevel).setLevelId(levelId).setViewPointId(viewPointId).setStartDate(startDate)
.setSpanType(spanType.getValue()).setSpanTypeDesc(spanTypeDesc).setCallType(callType).setApplicationId(applicationId).setUserId(userId).build().toByteArray();
}
@Override
public AbstractDataSerializable convertData(byte[] data) {
RequestSpan requestSpan = new RequestSpan();
try {
TraceProtocol.RequestSpan requestSpanByte = TraceProtocol.RequestSpan.parseFrom(data);
requestSpan.setTraceId(requestSpanByte.getTraceId());
requestSpan.setParentLevel(requestSpanByte.getParentLevel());
requestSpan.setLevelId(requestSpanByte.getLevelId());
requestSpan.setApplicationId(requestSpanByte.getApplicationId());
requestSpan.setCallType(requestSpanByte.getCallType());
requestSpan.setSpanType(SpanType.convert(requestSpanByte.getSpanType()));
requestSpan.setSpanTypeDesc(requestSpanByte.getSpanTypeDesc());
requestSpan.setStartDate(requestSpanByte.getStartDate());
requestSpan.setUserId(requestSpanByte.getUserId());
requestSpan.setViewPointId(requestSpanByte.getViewPointId());
} catch (InvalidProtocolBufferException e) {
return null;
}
return requestSpan;
}
@Override
public boolean isNull() {
return false;
}
public static class RequestSpanBuilder {
private RequestSpan ackSpan;
private RequestSpanBuilder(Span span) {
ackSpan = new RequestSpan(span);
}
public static RequestSpanBuilder newBuilder(Span span) {
return new RequestSpanBuilder(span);
}
public RequestSpanBuilder applicationId(String applicationId) {
ackSpan.applicationId = applicationId;
return this;
}
public RequestSpanBuilder callType(String callType) {
ackSpan.callType = callType;
return this;
}
public RequestSpanBuilder spanTypeDesc(String spanTypeDesc) {
ackSpan.spanTypeDesc = spanTypeDesc;
return this;
}
public RequestSpanBuilder userId(String userId) {
ackSpan.userId = userId;
return this;
}
public RequestSpan build() {
return ackSpan;
}
public RequestSpanBuilder viewPoint(String viewPoint) {
ackSpan.viewPointId = viewPoint;
return this;
}
}
}

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

@ -1,65 +0,0 @@
package com.ai.cloud.skywalking.protocol;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import com.ai.cloud.skywalking.protocol.common.NullableClass;
import com.ai.cloud.skywalking.protocol.exception.SerializableDataTypeRegisterException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
/**
* Created by wusheng on 16/7/4.
*/
public class SerializableDataTypeRegister {
private static Map<Integer, Class<?>> DATA_TYPE_MAPPING_CLASS = new HashMap<Integer, Class<?>>();
private static Map<Class<?>, Integer> CLASS_MAPPING_DATA_TYPE = new HashMap<Class<?>, Integer>();
public static void init(Integer dataType, Class<?> clazz) {
if (DATA_TYPE_MAPPING_CLASS.containsKey(dataType)) {
if (!DATA_TYPE_MAPPING_CLASS.get(dataType).equals(clazz)) {
throw new SerializableDataTypeRegisterException("dataType=" + dataType + " has been registered to " + DATA_TYPE_MAPPING_CLASS.get(dataType));
} else {
return;
}
}
DATA_TYPE_MAPPING_CLASS.put(dataType, clazz);
CLASS_MAPPING_DATA_TYPE.put(clazz, dataType);
}
public static boolean isTypeAndClassMatch(Integer dataType, Class<?> clazz) {
if (DATA_TYPE_MAPPING_CLASS.containsKey(dataType)) {
return true;
} else {
return false;
}
}
public static ISerializable findSerializableClassAndSerialize(byte[] data) {
Set<Class<?>> registerSerializableClasses = CLASS_MAPPING_DATA_TYPE.keySet();
for (Class<?> serializableClass : registerSerializableClasses) {
try {
ISerializable result = ((ISerializable) serializableClass.newInstance());
//TODO
NullableClass nullableClass = result.convert2Object(data);
if (!nullableClass.isNull()){
}
} catch (Exception e) {
}
}
return null;
}
public static Integer getType(Class<?> clazz) {
if (CLASS_MAPPING_DATA_TYPE.containsKey(clazz)) {
return CLASS_MAPPING_DATA_TYPE.get(clazz);
} else {
throw new SerializableDataTypeRegisterException("class " + clazz + " not found in SerializableDataTypeRegister.");
}
}
}

307
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java
View File

@ -4,209 +4,157 @@ import com.ai.cloud.skywalking.protocol.common.SpanType;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
public class Span extends SpanData {
public class Span {
private Logger logger = Logger.getLogger(Span.class.getName());
/**
* tid，调用链的全局唯一标识
*/
protected String traceId;
/**
* 当前调用链的上级描述<br/>
* 如当前序号为：0.1.0时，parentLevel=0.1
*/
protected String parentLevel;
/**
* 当前调用链的本机描述<br/>
* 如当前序号为：0.1.0时，levelId=0
*/
protected int levelId = 0;
public Span() {
/**
* 节点调用开始时间
*/
protected long startDate = System.currentTimeMillis();
/**
* 节点调用的状态<br/>
* 0：成功<br/>
* 1：异常<br/>
* 异常判断原则：代码产生exception，并且此exception不在忽略列表中
*/
protected byte statusCode = 0;
/**
* 节点调用的错误堆栈<br/>
* 堆栈以JAVA的exception为主要判断依据
*/
protected String exceptionStack;
/**
* 节点的状态<br/>
* 不参与序列化
*/
protected boolean isInvalidate = false;
/**
* 节点调用过程中的业务字段<br/>
* 如：业务系统设置的订单号，SQL语句等
*/
protected Map<String, String> parameters = new HashMap<String, String>();
/**
* 节点类型<br/>
* 如：RPC Client,RPC Server,Local
*/
private SpanType spanType = SpanType.LOCAL;
public Span(String traceId) {
this.traceId = traceId;
}
public Span(String traceId, String applicationID, String userId) {
public Span(String traceId, String parentLevel, int levelId) {
this.traceId = traceId;
this.applicationId = applicationID;
this.userId = userId;
}
public Span(String traceId, String parentLevelId, int levelId,
String applicationID, String userId) {
this.traceId = traceId;
this.applicationId = applicationID;
this.parentLevel = parentLevelId;
this.userId = userId;
this.parentLevel = parentLevel;
this.levelId = levelId;
}
public Span(String originData) {
String[] fieldValues = originData.split(SPAN_FIELD_SEPARATOR);
int index = 0;
while (this.setValueByIndex(fieldValues, index)) {
index++;
}
this.originData = originData;
public String getTraceId() {
return traceId;
}
private boolean setValueByIndex(String[] fieldValues, int index) {
if (fieldValues.length > index) {
switch (index) {
case 0:
traceId = fieldValues[0].trim();
break;
case 1:
parentLevel = fieldValues[1].trim();
break;
case 2:
levelId = Integer.valueOf(fieldValues[2]);
break;
case 3:
viewPointId = fieldValues[3].trim();
break;
case 4:
startDate = Long.valueOf(fieldValues[4]);
break;
case 5:
cost = Long.parseLong(fieldValues[5]);
break;
case 6:
address = fieldValues[6].trim();
break;
case 7:
statusCode = Byte.valueOf(fieldValues[7].trim());
break;
case 8:
exceptionStack = fieldValues[8].trim().replaceAll(
NEW_LINE_PLACEHOLDER, OS_NEW_LINE);
break;
case 9:
spanTypeDesc = fieldValues[9];
break;
case 10:
spanType = SpanType.convert(fieldValues[10]);
break;
case 11:
businessKey = fieldValues[11].trim().replaceAll(
NEW_LINE_PLACEHOLDER, OS_NEW_LINE);
break;
case 12:
processNo = fieldValues[12].trim();
break;
case 13:
applicationId = fieldValues[13].trim();
break;
case 14:
userId = fieldValues[14].trim();
break;
case 15:
callType = fieldValues[15].trim();
break;
default:
return false;
}
return true;
} else {
return false;
}
public String getParentLevel() {
return parentLevel;
}
@Override
public String toString() {
StringBuilder toStringValue = new StringBuilder();
toStringValue.append(traceId + SPAN_FIELD_SEPARATOR);
if (isNonBlank(parentLevel)) {
toStringValue.append(parentLevel + SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
toStringValue.append(levelId + SPAN_FIELD_SEPARATOR);
if (isNonBlank(viewPointId)) {
toStringValue.append(generateViewPointBySpanType() + SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
toStringValue.append(startDate + SPAN_FIELD_SEPARATOR);
toStringValue.append(cost + SPAN_FIELD_SEPARATOR);
if (isNonBlank(address)) {
toStringValue.append(address + SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
toStringValue.append(statusCode + SPAN_FIELD_SEPARATOR);
if (isNonBlank(exceptionStack)) {
// 换行符在各个系统中表现不一致，
// windows平台的换行符为/r/n
// linux平台的换行符为/n
toStringValue.append(exceptionStack.replaceAll(
WINDOWS_OS_NEW_LINE_REDUNDANT_CHAR, "").replaceAll(
OS_NEW_LINE, NEW_LINE_PLACEHOLDER)
+ SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
toStringValue.append(spanTypeDesc + SPAN_FIELD_SEPARATOR);
toStringValue.append(spanType.getValue() + SPAN_FIELD_SEPARATOR);
if (isNonBlank(businessKey)) {
// 换行符在各个系统中表现不一致，
// windows平台的换行符为/r/n
// linux平台的换行符为/n
toStringValue.append(businessKey.replaceAll(
WINDOWS_OS_NEW_LINE_REDUNDANT_CHAR, "").replaceAll(
OS_NEW_LINE, NEW_LINE_PLACEHOLDER)
+ SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
if (isNonBlank(processNo)) {
toStringValue.append(processNo + SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
if (isNonBlank(applicationId)) {
toStringValue.append(applicationId + SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
if (isNonBlank(userId)) {
toStringValue.append(userId + SPAN_FIELD_SEPARATOR);
} else {
toStringValue.append(" " + SPAN_FIELD_SEPARATOR);
}
toStringValue.append(callType);
return toStringValue.toString();
public void setParentLevel(String parentLevel) {
this.parentLevel = parentLevel;
}
private String generateViewPointBySpanType() {
public int getLevelId() {
return levelId;
}
public void setLevelId(int levelId) {
this.levelId = levelId;
}
public long getStartDate() {
return startDate;
}
public void setStartDate(long startDate) {
this.startDate = startDate;
}
public boolean isInvalidate() {
return isInvalidate;
}
public byte getStatusCode() {
return statusCode;
}
public String getExceptionStack() {
return exceptionStack;
}
public void setExceptionStack(String exceptionStack) {
this.exceptionStack = exceptionStack;
}
public Map<String, String> getParameters() {
return parameters;
}
public void setParameters(Map<String, String> parameters) {
this.parameters = parameters;
}
public void setInvalidate(boolean invalidate) {
isInvalidate = invalidate;
}
public boolean isRPCClientSpan() {
if (spanType == SpanType.RPC_CLIENT) {
viewPointId = "RPC Client : " + viewPointId;
} else if (spanType == SpanType.RPC_SERVER) {
viewPointId = "RPC Server : " + viewPointId;
return true;
}
return viewPointId;
return false;
}
protected boolean isNonBlank(String str) {
return str != null && str.length() > 0;
public void setSpanType(SpanType spanType) {
this.spanType = spanType;
}
public void handleException(Throwable e, Set<String> exclusiveExceptionSet,
int maxExceptionStackLength) {
public SpanType getSpanType() {
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();
try {
buf = new ByteArrayOutputStream();
Throwable causeException = e;
while (expMessage.length() < maxExceptionStackLength && causeException != null) {
causeException.printStackTrace(new java.io.PrintWriter(buf,
true));
causeException.printStackTrace(new java.io.PrintWriter(buf, true));
expMessage.append(buf.toString());
causeException = causeException.getCause();
}
@ -215,8 +163,7 @@ public class Span extends SpanData {
try {
buf.close();
} catch (IOException ioe) {
logger.log(Level.ALL,
"Close exception stack input stream failed", ioe);
logger.log(Level.ALL, "Close exception stack input stream failed", ioe);
}
}
@ -232,15 +179,5 @@ public class Span extends SpanData {
}
}
public boolean isRPCClientSpan() {
if (this.spanType == SpanType.RPC_CLIENT) {
return true;
} else {
return false;
}
}
public void setIsInvalidate(boolean isInvalidate) {
this.isInvalidate = isInvalidate;
}
}

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

@ -1,240 +0,0 @@
package com.ai.cloud.skywalking.protocol;
import com.ai.cloud.skywalking.protocol.common.CallType;
import com.ai.cloud.skywalking.protocol.common.SpanType;
import java.util.HashMap;
import java.util.Map;
public abstract class SpanData {
/**
* Span在序列中，各字段间的分隔符
*/
protected static final String SPAN_FIELD_SEPARATOR = "@~";
/**
* Span在序列化中，新的换行符
*/
protected static final String NEW_LINE_PLACEHOLDER = "#~";
/**
* 字符串中的换行符
*/
protected static final String OS_NEW_LINE = "\n";
/**
* windows操作系统的多余换行字符
*/
protected static final String WINDOWS_OS_NEW_LINE_REDUNDANT_CHAR = "\r";
/**
* tid，调用链的全局唯一标识
*/
protected String traceId;
/**
* 当前调用链的上级描述<br/>
* 如当前序号为：0.1.0时，parentLevel=0.1
*/
protected String parentLevel;
/**
* 当前调用链的本机描述<br/>
* 如当前序号为：0.1.0时，levelId=0
*/
protected int levelId = 0;
/**
* 调用链中单个节点的入口描述<br/>
* 如：java方法名，调用的RPC地址等等
*/
protected String viewPointId = "";
/**
* 节点调用开始时间
*/
protected long startDate = System.currentTimeMillis();
/**
* 节点调用花费时间
*/
protected long cost = 0L;
/**
* 节点调用的发生机器描述<br/>
* 包含机器名 + IP地址
*/
protected String address = "";
/**
* 节点调用的状态<br/>
* 0：成功<br/>
* 1：异常<br/>
* 异常判断原则：代码产生exception，并且此exception不在忽略列表中
*/
protected byte statusCode = 0;
/**
* 节点调用的错误堆栈<br/>
* 堆栈以JAVA的exception为主要判断依据
*/
protected String exceptionStack;
/**
* 节点类型描述<br/>
* 已字符串的形式描述<br/>
* 如：java,dubbo等
*/
protected String spanTypeDesc = "";
/**
* 节点调用类型描述<br/>
* @see CallType
*/
protected String callType = "";
/**
* 节点的状态<br/>
* 不参与序列化
*/
protected boolean isInvalidate = false;
/**
* 节点分布式类型<br/>
* 本地调用 / RPC服务端 / RPC客户端
*/
protected SpanType spanType = SpanType.LOCAL;
/**
* 节点调用过程中的业务字段<br/>
* 如：业务系统设置的订单号，SQL语句等
*/
protected Map<String, String> parameters = new HashMap<String, String>();
/**
* 节点调用的所在进程号
*/
protected String processNo = "";
/**
* 节点调用所在的系统逻辑名称<br/>
* 由授权文件指定
*/
protected String applicationId = "";
/**
* 反序列化时，存储序列化前的字符串原文
*/
protected String originData = "";
/**
* 用户id<br/>
* 由授权文件指定
*/
protected String userId;
public String getTraceId() {
return traceId;
}
public String getParentLevel() {
return parentLevel;
}
public void setParentLevel(String parentLevel) {
this.parentLevel = parentLevel;
}
public int getLevelId() {
return levelId;
}
public void setLevelId(int levelId) {
this.levelId = levelId;
}
public String getViewPointId() {
return viewPointId;
}
public void setViewPointId(String viewPointId) {
this.viewPointId = viewPointId;
}
public long getStartDate() {
return startDate;
}
public void setStartDate(long startDate) {
this.startDate = startDate;
}
public void setCost(long cost) {
this.cost = cost;
}
public void setAddress(String address) {
this.address = address;
}
public String getSpanTypeDesc() {
return spanTypeDesc;
}
public void setSpanTypeDesc(String spanTypeDesc) {
this.spanTypeDesc = spanTypeDesc;
}
public SpanType getSpanType() {
return spanType;
}
public void setSpanType(SpanType spanType) {
this.spanType = spanType;
}
public boolean isInvalidate() {
return isInvalidate;
}
public void setProcessNo(String processNo) {
this.processNo = processNo;
}
public String getOriginData() {
return originData;
}
public long getCost() {
return cost;
}
public String getAddress() {
return address;
}
public byte getStatusCode() {
return statusCode;
}
public String getExceptionStack() {
return exceptionStack;
}
public void setExceptionStack(String exceptionStack) {
this.exceptionStack = exceptionStack;
}
public Map<String, String> getParameters() {
return parameters;
}
public void setParameters(Map<String, String> parameters) {
this.parameters = parameters;
}
public String getProcessNo() {
return processNo;
}
public String getApplicationId() {
return applicationId;
}
public String getUserId() {
return userId;
}
public void setCallType(String callType) {
this.callType = callType;
}
public String getCallType() {
return callType;
}
}

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

@ -1,9 +1,10 @@
package com.ai.cloud.skywalking.protocol.common;
import com.ai.cloud.skywalking.protocol.NullClass;
import com.ai.cloud.skywalking.protocol.SerializableDataTypeRegister;
import com.ai.cloud.skywalking.serialize.SerializedFactory;
import com.ai.cloud.skywalking.util.IntegerAssist;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
@ -13,14 +14,12 @@ import java.util.Set;
public abstract class AbstractDataSerializable implements ISerializable, NullableClass {
private static Set<Integer> DATA_TYPE_SCOPE = new HashSet<Integer>();
public AbstractDataSerializable(){
SerializableDataTypeRegister.init(getDataType(), this.getClass());
}
public abstract int getDataType();
public abstract byte[] getData();
public abstract AbstractDataSerializable convertData(byte[] data);
/**
* 消息包结构:
* 4位消息体类型
@ -30,21 +29,31 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl
*/
@Override
public byte[] convert2Bytes() {
byte[] type = IntegerAssist.intToBytes(SerializableDataTypeRegister.getType(this.getClass()));
byte[] messagePackage = new byte[4 + getData().length];
appendingDataText(messagePackage);
appendingDataLength(messagePackage);
return messagePackage;
}
//TODO:消息包 = 4位
return getData();
private void appendingDataLength(byte[] dataByte) {
byte[] type = IntegerAssist.intToBytes(this.getDataType());
System.arraycopy(type, 0, dataByte, 0, type.length);
}
private byte[] appendingDataText(byte[] dataByte) {
System.arraycopy(getData(), 0, dataByte, 4, dataByte.length);
return dataByte;
}
@Override
public NullableClass convert2Object(byte[] data) {
// TODO:data的前4位转成type;
int dataType = 1;
if(!SerializableDataTypeRegister.isTypeAndClassMatch(dataType, this.getClass())){
int dataType = IntegerAssist.bytesToInt(data, 0);
if (!SerializedFactory.isCanSerialized(dataType)) {
return new NullClass();
}
// TODO: 反序列化
return null;
return this.convertData(Arrays.copyOfRange(data,4, data.length));
}

25
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/SpanType.java
View File

@ -7,26 +7,31 @@ import com.ai.cloud.skywalking.protocol.exception.SpanTypeCannotConvertException
*/
public enum SpanType {
LOCAL((byte) 1), RPC_CLIENT((byte) 2), RPC_SERVER((byte) 4);
LOCAL(1),
RPC_CLIENT(2),
RPC_SERVER(4);
private byte value;
private int value;
SpanType(byte value) {
SpanType(int value) {
this.value = value;
}
public static SpanType convert(String spanTypeValue) {
switch (Byte.valueOf(spanTypeValue)){
case 1 : return LOCAL;
case 2 : return RPC_CLIENT;
case 3 : return RPC_SERVER;
public static SpanType convert(int spanTypeValue) {
switch (spanTypeValue) {
case 1:
return LOCAL;
case 2:
return RPC_CLIENT;
case 3:
return RPC_SERVER;
default:
throw new SpanTypeCannotConvertException(spanTypeValue);
throw new SpanTypeCannotConvertException(spanTypeValue + "");
}
}
public byte getValue() {
public int getValue() {
return value;
}
}

45
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/serialize/SerializedFactory.java Normal file
View File

@ -0,0 +1,45 @@
package com.ai.cloud.skywalking.serialize;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.common.NullableClass;
import com.ai.cloud.skywalking.util.IntegerAssist;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.HashMap;
import java.util.Map;
import java.util.ServiceLoader;
public class SerializedFactory {
private static Logger logger = LogManager.getLogger(SerializedFactory.class);
public static Map<Integer, AbstractDataSerializable> serializableMap = new HashMap<Integer, AbstractDataSerializable>();
static {
ServiceLoader<AbstractDataSerializable> loaders = ServiceLoader.load(AbstractDataSerializable.class);
for (AbstractDataSerializable serializable : loaders) {
serializableMap.put(serializable.getDataType(), serializable);
}
}
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;
}
}
return null;
}
public static byte[] serialize(AbstractDataSerializable dataSerializable) {
return dataSerializable.convert2Bytes();
}
public static boolean isCanSerialized(int dataType) {
return serializableMap.get(dataType) != null ? true : false;
}
}

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

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

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

@ -1,6 +1,5 @@
package com.ai.cloud.skywalking.util;
import com.ai.cloud.skywalking.protocol.SerializableDataTypeRegister;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import java.util.ArrayList;
@ -16,16 +15,16 @@ public class TransportPackager {
return dataPackage;
}
public static List<ISerializable> unpack(byte[] dataPackage) {
public static List<byte[]> unpack(byte[] dataPackage) {
if (validateCheckSum(dataPackage)) {
return unpackDataText(dataPackage);
} else {
return null;
return new ArrayList<byte[]>();
}
}
private static List<ISerializable> unpackDataText(byte[] dataPackage) {
List<ISerializable> serializeData = new ArrayList<ISerializable>();
private static List<byte[]> unpackDataText(byte[] dataPackage) {
List<byte[]> serializeData = new ArrayList<byte[]>();
int currentLength = 0;
while (true) {
//读取长度
@ -34,10 +33,7 @@ public class TransportPackager {
byte[] data = new byte[dataLength];
System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength);
//
ISerializable data1 = SerializableDataTypeRegister.findSerializableClassAndSerialize(data);
if (data1 != null) {
serializeData.add(data1);
}
serializeData.add(data);
currentLength = 4 + dataLength;
if (currentLength >= dataPackage.length) {
break;

13
skywalking-collector/skywalking-protocol/src/main/proto/AckSpanProtobuf.proto
View File

@ -1,13 +0,0 @@
syntax="proto2";
option java_package = "com.ai.cloud.skywalking.protocol.proto";
message AckSpan{
required string traceId = 1;
optional string parentLevel = 2;
optional int32 levelId = 3;
required int64 cost = 4;
required int32 statusCode = 5;
optional string exceptionStack = 6;
}

26
skywalking-collector/skywalking-protocol/src/main/proto/TraceProtocol.proto Normal file
View File

@ -0,0 +1,26 @@
syntax="proto2";
option java_package = "com.ai.cloud.skywalking.protocol.proto";
message AckSpan{
required string traceId = 1;
optional string parentLevel = 2;
required int32 levelId = 3;
required int64 cost = 4;
required int32 statusCode = 5;
optional string exceptionStack = 6;
}
message RequestSpan {
required string traceId = 1;
optional string parentLevel = 2;
required int32 levelId = 3;
required string viewPointId = 4;
required int64 startDate = 5;
required string spanTypeDesc = 6;
required string callType = 7;
required uint32 spanType = 8;
required string applicationId = 9;
required string userId = 10;
}

2
skywalking-collector/skywalking-protocol/src/main/resources/META-INF/services/com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable Normal file
View File

@ -0,0 +1,2 @@
com.ai.cloud.skywalking.protocol.AckSpan
com.ai.cloud.skywalking.protocol.RequestSpan

32
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java
View File

@ -1,11 +1,12 @@
package com.ai.cloud.skywalking.reciever;
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.selfexamination.ServerHealthCollector;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
@ -14,19 +15,11 @@ import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.io.IOException;
import java.util.Properties;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
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.persistance.PersistenceThreadLauncher;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import java.io.IOException;
import java.util.Properties;
public class CollectionServer {
@ -42,15 +35,12 @@ 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 {
ChannelPipeline p = ch.pipeline();
p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0,4));
p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
p.addLast("frameEncoder", new LengthFieldPrepender(4));
p.addLast("decoder", new ByteArrayDecoder());
p.addLast("encoder", new ByteArrayEncoder());
@ -71,8 +61,6 @@ 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.....");

41
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/AppendEOFFlagThread.java
View File

@ -1,41 +0,0 @@
package com.ai.cloud.skywalking.reciever.buffer;
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;
class AppendEOFFlagThread extends Thread {
private Logger logger = LogManager.getLogger(AppendEOFFlagThread.class);
private File[] dataBufferFiles;
public AppendEOFFlagThread(File[] dataBufferFiles) {
super("AppendEOFFlagThread");
this.dataBufferFiles = dataBufferFiles;
}
@Override
public void run() {
BufferedWriter bufferedWriter = null;
for (File file : dataBufferFiles) {
try {
logger.info("Add EOF flags to unprocessed data file[{}]", file.getName());
bufferedWriter = new BufferedWriter(new FileWriter(new File(file.getParent(), file.getName()), true));
bufferedWriter.write("EOF\n");
} catch (IOException e) {
logger.info("Add EOF flags to the unprocessed data file failed.", e);
} finally {
try {
bufferedWriter.flush();
bufferedWriter.close();
} catch (IOException e) {
logger.error("Flush data file failed", e);
}
}
}
}
}

154
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java
View File

@ -1,157 +1,77 @@
package com.ai.cloud.skywalking.reciever.buffer;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.DATA_BUFFER_FILE_PARENT_DIRECTORY;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.DATA_CONFLICT_WAIT_TIME;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.DATA_FILE_MAX_LENGTH;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.FLUSH_NUMBER_OF_CACHE;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.MAX_WAIT_TIME;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.PER_THREAD_MAX_BUFFER_NUMBER;
import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.WRITE_DATA_FAILURE_RETRY_INTERVAL;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.UUID;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.processor.AbstractSpanProcessor;
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 com.ai.cloud.skywalking.serialize.SerializedFactory;
import com.ai.cloud.skywalking.util.AtomicRangeInteger;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
import com.ai.cloud.skywalking.util.AtomicRangeInteger;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
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;
public class DataBufferThread extends Thread {
private Logger logger = LogManager.getLogger(DataBufferThread.class);
private byte[][] data = new byte[PER_THREAD_MAX_BUFFER_NUMBER][];
private File file;
private FileOutputStream outputStream;
private AtomicRangeInteger index = new AtomicRangeInteger(0, PER_THREAD_MAX_BUFFER_NUMBER);
private Logger logger = LogManager.getLogger(DataBufferThread.class);
private byte[][] data = new byte[PER_THREAD_MAX_BUFFER_NUMBER][];
private AtomicRangeInteger index = new AtomicRangeInteger(0, PER_THREAD_MAX_BUFFER_NUMBER);
public DataBufferThread(int threadIdx) {
super("DataBufferThread_" + threadIdx);
try {
file = new File(DATA_BUFFER_FILE_PARENT_DIRECTORY, getFileName());
if (file.exists()) {
file.createNewFile();
}
if (logger.isDebugEnabled()) {
logger.debug("Create buffer data file {}.", file.getName());
}
outputStream = new FileOutputStream(file, true);
} catch (FileNotFoundException e) {
logger.error("Data cache file cannot be created or written, please check the file system.", e);
System.exit(-1);
} catch (IOException e) {
logger.error("Data cache file cannot be created or written, please check the file system.", e);
System.exit(-1);
}
super("DataBufferThread_" + threadIdx);
}
@Override
public void run() {
boolean isWriteFailure;
int index = 0;
Map<Integer, List<AbstractDataSerializable>> serializeObjects;
while (true) {
boolean hasData2Flush = false;
serializeObjects = new HashMap<Integer, List<AbstractDataSerializable>>();
for (int i = 0; i < data.length; i++) {
if (data[i] == null) {
continue;
}
hasData2Flush = true;
isWriteFailure = true;
while (isWriteFailure) {
try {
outputStream.write(data[i]);
outputStream.write("\n".getBytes());
isWriteFailure = false;
} catch (IOException e) {
logger.error("Write buffer data failed.", e);
try {
Thread.sleep(WRITE_DATA_FAILURE_RETRY_INTERVAL);
} catch (InterruptedException e1) {
logger.error("Failure sleep.", e);
}
}
}
if (index++ > FLUSH_NUMBER_OF_CACHE) {
try {
outputStream.flush();
ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer flush data to local file:" + file.getName());
} catch (IOException e) {
logger.error("Flush buffer data failed.", e);
} finally {
index = 0;
}
AbstractDataSerializable serializeData = SerializedFactory.unSerialize(data[i]);
List<AbstractDataSerializable> hasBeenSerializedObjects = serializeObjects.get(serializeData.getDataType());
if (hasBeenSerializedObjects == null) {
serializeObjects.put(serializeData.getDataType(), new ArrayList<AbstractDataSerializable>());
}
serializeObjects.get(serializeData.getDataType()).add(serializeData);
data[i] = null;
}
if (hasData2Flush){
try {
outputStream.flush();
ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer flush data to local file:" + file.getName());
} catch (IOException e) {
logger.error("Flush buffer data failed.", e);
for (Map.Entry<Integer, List<AbstractDataSerializable>> entry : serializeObjects.entrySet()) {
AbstractSpanProcessor processor = ProcessorFactory.chooseProcessor(entry.getKey());
if (processor != null) {
processor.process(entry.getValue());
}
}
if (file.length() > DATA_FILE_MAX_LENGTH) {
switchFile();
try {
Thread.sleep(Config.Buffer.MAX_WAIT_TIME);
} catch (InterruptedException e) {
logger.error("Failed to sleep.", e);
}
if (!hasData2Flush) {
try {
Thread.sleep(MAX_WAIT_TIME);
} catch (InterruptedException e) {
logger.error("Failure sleep.", e);
}
}
}
}
private String getFileName() {
return System.currentTimeMillis() + "-" + UUID.randomUUID().toString().replaceAll("-", "");
}
private void switchFile() {
String fileName = getFileName();
try {
outputStream.write("EOF\n".getBytes());
outputStream.flush();
} catch (IOException e) {
logger.error("Write eof to cache data file.", e);
} finally{
try {
outputStream.close();
} catch (IOException e) {
logger.error("close cache data failed.", e);
}
ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer close local file:" + file.getName());
}
logger.debug("Begin to switch the data file to {}.", fileName);
try {
file = new File(DATA_BUFFER_FILE_PARENT_DIRECTORY, fileName);
outputStream = new FileOutputStream(file, true);
ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer open new local file:" + file.getName());
} catch (IOException e) {
ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.ERROR, "DataBuffer open new local file failure.");
logger.error("Switch data file failed.", e);
}
}
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);

42
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java
View File

@ -1,22 +1,16 @@
package com.ai.cloud.skywalking.reciever.buffer;
import static com.ai.cloud.skywalking.reciever.conf.Config.Persistence.MAX_APPEND_EOF_FLAGS_THREAD_NUMBER;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.commons.io.comparator.NameFileComparator;
import com.ai.cloud.skywalking.reciever.conf.Config;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.ai.cloud.skywalking.reciever.conf.Config;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
public class DataBufferThreadContainer {
private static Logger logger = LogManager.getLogger(DataBufferThreadContainer.class);
private static Logger logger = LogManager.getLogger(DataBufferThreadContainer.class);
private static List<DataBufferThread> buffers = new ArrayList<DataBufferThread>();
private DataBufferThreadContainer() {
@ -30,32 +24,6 @@ public class DataBufferThreadContainer {
}
public static void init() {
logger.info("Add EOF flags to the unprocessed data file last time.");
File parentDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIRECTORY);
// 判断数据缓存文件是否存在，如果不存在，则创建
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;
while (true) {
if (end + step >= dataFileList.length) {
new AppendEOFFlagThread(Arrays.copyOfRange(dataFileList, start, dataFileList.length)).start();
break;
}
end += step;
new AppendEOFFlagThread(Arrays.copyOfRange(dataFileList, start, end)).start();
start = end;
logger.debug("start:" + start + "\tend:" + end);
}
}
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();

86
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java
View File

@ -5,17 +5,16 @@ public class Config {
// 采集服务配置类
public static class Server {
// 采集服务器的端口
public static int PORT = 34000;
public static int PORT = 34000;
// 最大数据处理线程数量
public static int MAX_DEAL_DATA_THREAD_NUMBER = 3;
public static int MAX_DEAL_DATA_THREAD_NUMBER = 3;
// 异常数据的时间间隔
public static int FAILED_PACKAGE_WATCHING_TIME_WINDOW = 5 * 60;
// 时间间隔内最大异常数据次数
public static int MAX_WATCHING_FAILED_PACKAGE_SIZE = 200;
public static int MAX_WATCHING_FAILED_PACKAGE_SIZE = 200;
}
// 数据缓存配置类
public static class Buffer {
@ -28,59 +27,13 @@ public class Config {
// 数据冲突时等待时间(单位:毫秒)
public static long DATA_CONFLICT_WAIT_TIME = 10L;
// 数据缓存文件目录
public static String DATA_BUFFER_FILE_PARENT_DIRECTORY = "../data/buffer";
// 缓存数据文件最大长度(单位:byte)
public static int DATA_FILE_MAX_LENGTH = 30 * 1024 * 1024;
// 每次缓存数据写入失败，最大尝试时间
public static long WRITE_DATA_FAILURE_RETRY_INTERVAL =10 * 1000L;
//每次Flush的缓存数据的个数
public static int FLUSH_NUMBER_OF_CACHE = 30;
}
public static class DataPackage {
public static int MAX_DATA_PACKAGE = 1024 * 1024;
}
public static class Persistence {
// 定位文件时,每次读取偏移量跳过大小
public static int STEP_SIZE_FOR_LOCATING_FILE_OFFSET = 2048;
// 切换文件，等待时间
public static long SWITCH_FILE_WAIT_TIME = 5000L;
// 追加EOF标志位的线程数量
public static int MAX_APPEND_EOF_FLAGS_THREAD_NUMBER = 2;
// 每次存储的最大数量
public static int MAX_STORAGE_SIZE_PER_TIME = 1024 * 1024;
// 当读取到文件结束时等待时间
public static long READ_ENDING_FILE_MAX_WAITE_TIME = 500L;
}
public static class RegisterPersistence {
// 偏移量注册文件的目录
public static String REGISTER_FILE_PARENT_DIRECTORY = "../data/offset";
// 偏移量注册文件名
public static String REGISTER_FILE_NAME = "offset.txt";
// 偏移量注册备份文件名
public static String REGISTER_BAK_FILE_NAME = "offset.txt.bak";
// 偏移量写入文件等待周期
public static long OFFSET_WRITTEN_FILE_WAIT_CYCLE = 5000L;
}
public static class HBaseConfig {
public static String TABLE_NAME = "sw-call-chain";
public static String FAMILY_COLUMN_NAME = "call-chain";
public static String ZK_HOSTNAME;
@ -88,11 +41,13 @@ public class Config {
public static String CLIENT_PORT;
}
public static class StorageChain {
public static String STORAGE_TYPE = "hbase";
}
public static class Redis{
public static class Redis {
public static String REDIS_SERVER = "10.1.241.18:16379";
@ -103,26 +58,29 @@ public class Config {
public static int REDIS_MAX_TOTAL = 20;
}
public static class Alarm {
public static int ALARM_EXPIRE_SECONDS = 1000 * 60 * 90;
public static int ALARM_EXCEPTION_STACK_LENGTH = 300;
public static boolean ALARM_OFF_FLAG = false;
public static long ALARM_REDIS_INSPECTOR_INTERVAL = 5 * 1000L;
public static class Checker {
public static boolean TURN_ON_EXCEPTION_CHECKER = true;
public static boolean TURN_ON_EXECUTE_TIME_CHECKER = true;
public static boolean TURN_ON_EXCEPTION_CHECKER = true;
public static boolean TURN_ON_EXECUTE_TIME_CHECKER = true;
}
}
public static class HealthCollector {
// 默认健康检查上报时间
public static long REPORT_INTERVAL = 5 * 60 * 1000L;
// 默认健康检查上报时间
public static long REPORT_INTERVAL = 5 * 60 * 1000L;
}
}
}

15
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java
View File

@ -9,6 +9,7 @@ import io.netty.channel.SimpleChannelInboundHandler;
import redis.clients.jedis.Jedis;
import java.net.InetSocketAddress;
import java.util.List;
public class CollectionServerDataHandler extends SimpleChannelInboundHandler<byte[]> {
@ -16,12 +17,12 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler<byt
protected void channelRead0(ChannelHandlerContext ctx, byte[] msg) throws Exception {
Thread.currentThread().setName("ServerReceiver");
// 当接受到这条消息的是空，则忽略
if (msg != null && msg.length >= 0 && msg.length < Config.DataPackage.MAX_DATA_PACKAGE) {
if (msg != null && msg.length >= 0) {
byte[] data = TransportPackager.unpack(msg);
List<byte[]> byteSerializeObjects = TransportPackager.unpack(msg);
if (data != null) {
DataBufferThreadContainer.getDataBufferThread().saveTemporarily(data);
if (byteSerializeObjects.size() > 0) {
cacheSerializeObjects(byteSerializeObjects);
} else {
// 处理错误包
dealFailedPackage(ctx);
@ -29,6 +30,12 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler<byt
}
}
private void cacheSerializeObjects(List<byte[]> byteSerializeObjects) {
for (byte[] byteSerializeObject : byteSerializeObjects) {
DataBufferThreadContainer.getDataBufferThread().saveTemporarily(byteSerializeObject);
}
}
private void dealFailedPackage(ChannelHandlerContext ctx) {
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().localAddress();
String key = ctx.name() + "-" + socketAddress.getHostName() + ":" + socketAddress.getPort();

76
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/FileRegisterEntry.java
View File

@ -1,76 +0,0 @@
package com.ai.cloud.skywalking.reciever.persistance;
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;
}
}

151
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/MemoryRegister.java
View File

@ -1,151 +0,0 @@
package com.ai.cloud.skywalking.reciever.persistance;
import com.ai.cloud.skywalking.reciever.conf.Config;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.*;
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() {
BufferedReader reader;
// 在处理数据之前需要初始化处理文件的处理状态
//去掉entries中无法与缓存数据文件匹配的文件
File parentDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIRECTORY);
//上次未处理的缓存数据文件，entries内的数据主要以缓存
List<String> bufferFileNameList = Arrays.asList(parentDir.list());
try {
// 读取offset文件
file = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_FILE_NAME);
// offset File不存在
if (!file.exists()) {
File offsetBackUpFile = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_BAK_FILE_NAME);
// offset备份文件存在
if (offsetBackUpFile.exists()) {
reader = new BufferedReader(new FileReader(offsetBackUpFile));
String offsetData;
while ((offsetData = reader.readLine()) != null && !"EOF".equals(offsetData)) {
String[] ss = offsetData.split("\t");
if (bufferFileNameList.contains(ss[0])) {
entries.put(ss[0], new FileRegisterEntry(ss[0], Integer.valueOf(ss[1]), FileRegisterEntry.FileRegisterEntryStatus.UNREGISTER));
}
}
}
// 创建offset文件
file.createNewFile();
} else {
// 如果存在
reader = new BufferedReader(new FileReader(file));
String offsetData;
while ((offsetData = reader.readLine()) != null && !"EOF".equals(offsetData)) {
try {
String[] ss = offsetData.split("\t");
if (bufferFileNameList.contains(ss[0])) {
entries.put(ss[0], new FileRegisterEntry(ss[0], Integer.valueOf(ss[1]), FileRegisterEntry.FileRegisterEntryStatus.UNREGISTER));
}
} catch (Exception e) {
continue;
}
}
}
} 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);
}
}
}

199
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java
View File

@ -1,199 +0,0 @@
package com.ai.cloud.skywalking.reciever.persistance;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
import com.ai.cloud.skywalking.reciever.storage.StorageChainController;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.comparator.NameFileComparator;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.*;
import static com.ai.cloud.skywalking.reciever.conf.Config.Persistence.*;
public class PersistenceThread extends Thread {
private Logger logger = LogManager.getLogger(PersistenceThread.class);
PersistenceThread(int threadIdx) {
super("PersistenceThread_" + threadIdx);
}
@Override
public void run() {
File file1 = null;
BufferedReader bufferedReader = null;
int offset;
while (true) {
try {
file1 = getDataFiles();
if (file1 == null) {
try {
Thread.sleep(SWITCH_FILE_WAIT_TIME);
} catch (InterruptedException e) {
logger.error("Failure sleep", e);
}
continue;
}
bufferedReader = new BufferedReader(new FileReader(file1));
offset = moveOffSet(file1, bufferedReader);
if (logger.isDebugEnabled()) {
logger.debug("Get file[{}] offset [{}]", file1.getName(),
offset);
}
StringBuilder stringBuilder = new StringBuilder(
MAX_STORAGE_SIZE_PER_TIME);
String tmpData;
while (true) {
tmpData = bufferedReader.readLine();
// 文件结束
if (tmpData == null) {
if (stringBuilder != null && stringBuilder.length() > 0) {
MemoryRegister.instance().updateOffSet(
file1.getName(), offset);
StorageChainController.doStorage(stringBuilder
.toString());
stringBuilder.delete(0, stringBuilder.length());
}
try {
Thread.sleep(READ_ENDING_FILE_MAX_WAITE_TIME);
} catch (InterruptedException e) {
logger.error("Sleep failed", e);
}
continue;
}
// 文件读入/n字符串
if (tmpData.length() <= 0) {
// 加上回车的字符串长度
offset += 1;
continue;
}
ServerHealthCollector.getCurrentHeathReading(null)
.updateData(
ServerHeathReading.INFO,
"read " + tmpData.length()
+ " chars from local file:"
+ file1.getName());
if ("EOF".equals(tmpData)) {
if (stringBuilder != null && stringBuilder.length() > 0) {
StorageChainController.doStorage(stringBuilder
.toString());
}
bufferedReader.close();
logger.info(
"Data in file[{}] has been successfully processed",
file1.getName());
boolean deleteSuccess = false;
while (!deleteSuccess) {
deleteSuccess = FileUtils.deleteQuietly(new File(
file1.getParent(), file1.getName()));
}
logger.info("Delete file[{}] {}", file1.getName(),
(deleteSuccess ? "success" : "failed"));
MemoryRegister.instance().removeEntry(file1.getName());
break;
}
if (stringBuilder.length() + tmpData.length() >= MAX_STORAGE_SIZE_PER_TIME) {
StorageChainController.doStorage(stringBuilder
.toString());
stringBuilder.delete(0, stringBuilder.length());
MemoryRegister.instance().updateOffSet(file1.getName(),
offset);
}
stringBuilder.append(tmpData);
// 加上回车的字符串长度
offset += tmpData.length() + 1;
}
} catch (FileNotFoundException e) {
logger.error("The data file could not be found.", e);
ServerHealthCollector.getCurrentHeathReading(null).updateData(
ServerHeathReading.ERROR, e.getMessage());
} catch (IOException e) {
logger.error("The data file I/O exception.", e);
ServerHealthCollector.getCurrentHeathReading(null).updateData(
ServerHeathReading.ERROR, e.getMessage());
} catch (Throwable t) {
logger.error(t.getMessage(), t);
ServerHealthCollector.getCurrentHeathReading(null).updateData(
ServerHeathReading.ERROR, t.getMessage());
} finally {
try{
if (file1 != null) {
MemoryRegister.instance().unRegister(file1.getName());
}
}catch (Throwable t) {
logger.error("unRegister file[{}] failure", file1.getName(), t);
}
try {
if (bufferedReader != null)
bufferedReader.close();
} catch (IOException e) {
logger.error("can't close data file", e);
}
}
try {
Thread.sleep(SWITCH_FILE_WAIT_TIME);
} catch (InterruptedException e) {
logger.error("Failure sleep.", e);
}
}
}
private int moveOffSet(File file1, BufferedReader bufferedReader)
throws IOException {
int offset = MemoryRegister.instance().getOffSet(file1.getName());
if (-1 == offset || offset == 0) {
offset = 0;
} else {
char[] cha = new char[STEP_SIZE_FOR_LOCATING_FILE_OFFSET];
int length = 0;
while (length + STEP_SIZE_FOR_LOCATING_FILE_OFFSET < offset) {
length += STEP_SIZE_FOR_LOCATING_FILE_OFFSET;
bufferedReader.read(cha);
}
bufferedReader.read(cha, 0, Math.abs(offset - length));
cha = null;
}
return offset;
}
private File getDataFiles() {
File file1 = null;
File parentDir = new File(
Config.Buffer.DATA_BUFFER_FILE_PARENT_DIRECTORY);
NameFileComparator sizeComparator = new NameFileComparator();
File[] dataFileList = sizeComparator.sort(parentDir.listFiles());
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;
}
}

13
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThreadLauncher.java
View File

@ -1,13 +0,0 @@
package com.ai.cloud.skywalking.reciever.persistance;
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();
}
}
}

95
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/RegisterPersistenceThread.java
View File

@ -1,95 +0,0 @@
package com.ai.cloud.skywalking.reciever.persistance;
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 BufferedWriter writer;
public RegisterPersistenceThread() {
super("RegisterPersistenceThread");
File offsetParentDir = new File(REGISTER_FILE_PARENT_DIRECTORY);
if (!offsetParentDir.exists()){
offsetParentDir.mkdirs();
}
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(OFFSET_WRITTEN_FILE_WAIT_CYCLE);
} catch (InterruptedException e) {
logger.error("Sleep failure", e);
}
try {
File file = new File(REGISTER_FILE_PARENT_DIRECTORY,
REGISTER_FILE_NAME);
File bakFile = new File(REGISTER_FILE_PARENT_DIRECTORY,
REGISTER_BAK_FILE_NAME);
// 先删除备份文件
if (bakFile.exists()) {
bakFile.delete();
}
// 将文件改名字
file.renameTo(bakFile);
//
if (!file.exists()) {
file.createNewFile();
}
Collection<FileRegisterEntry> fileRegisterEntries = MemoryRegister
.instance().getEntries();
try {
writer = new BufferedWriter(new FileWriter(file));
} catch (IOException e) {
logger.error("Write The offset file anomalies.");
}
for (FileRegisterEntry fileRegisterEntry : fileRegisterEntries) {
try {
writer.write(fileRegisterEntry.toString() + "\n");
} catch (IOException e) {
logger.error(
"Write file register entry to offset file failure", e);
}
}
try {
writer.write("EOF\n");
writer.flush();
} catch (IOException e) {
logger.error("Flush offset file failure", e);
} finally {
try {
writer.close();
} catch (IOException e) {
logger.error("close offset file failure", e);
}
}
} catch (IOException e) {
logger.error("Failed to back up offset file.", e);
}
ServerHealthCollector.getCurrentHeathReading(null).updateData(
ServerHeathReading.INFO, "flush memory register to file.");
}
}
}

53
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/AbstractSpanProcessor.java Normal file
View File

@ -0,0 +1,53 @@
package com.ai.cloud.skywalking.reciever.processor;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.processor.exception.HBaseInitFailedException;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.IOException;
import java.util.List;
public abstract class AbstractSpanProcessor implements IProcessor {
private static Logger logger = LogManager.getLogger(AbstractSpanProcessor.class);
private static Configuration configuration = null;
private static Connection connection;
static {
if (configuration == null) {
configuration = HBaseConfiguration.create();
if (Config.HBaseConfig.ZK_HOSTNAME == null || "".equals(Config.HBaseConfig.ZK_HOSTNAME)) {
logger.error("Miss HBase ZK quorum Configuration", new IllegalArgumentException("Miss HBase ZK quorum Configuration"));
System.exit(-1);
}
configuration.set("hbase.zookeeper.quorum", Config.HBaseConfig.ZK_HOSTNAME);
configuration.set("hbase.zookeeper.property.clientPort", Config.HBaseConfig.CLIENT_PORT);
}
try {
connection = ConnectionFactory.createConnection(configuration);
} catch (IOException e) {
ServerHealthCollector.getCurrentHeathReading("hbase").updateData(ServerHeathReading.ERROR, "connect to hbase failure.");
throw new HBaseInitFailedException("initHBaseClient failure", e);
}
}
@Override
public void process(List<AbstractDataSerializable> serializedObjects) {
doAlarm(serializedObjects);
doSaveHBase(connection, serializedObjects);
}
public abstract void doAlarm(List<AbstractDataSerializable> serializedObjects);
public abstract void doSaveHBase(Connection connection, List<AbstractDataSerializable> serializedObjects);
public abstract int getType();
}

51
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/AckSpanProcessor.java Normal file
View File

@ -0,0 +1,51 @@
package com.ai.cloud.skywalking.reciever.processor;
import com.ai.cloud.skywalking.protocol.AckSpan;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.util.HBaseUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import java.util.ArrayList;
import java.util.List;
import static com.ai.cloud.skywalking.reciever.util.SpanUtil.getTSBySpanTraceId;
public class AckSpanProcessor extends AbstractSpanProcessor {
@Override
public void doAlarm(List<AbstractDataSerializable> serializedObjects) {
}
@Override
public void doSaveHBase(Connection connection, List<AbstractDataSerializable> serializedObjects) {
List<Put> puts = new ArrayList<Put>();
// convert to put
String columnName;
for (AbstractDataSerializable serializedObject : serializedObjects) {
AckSpan ackSpan = (AckSpan) serializedObject;
Put put = new Put(Bytes.toBytes(ackSpan.getTraceId()), getTSBySpanTraceId(ackSpan.getTraceId()));
if (StringUtils.isEmpty(ackSpan.getParentLevel().trim())) {
columnName = ackSpan.getLevelId() + "";
} else {
columnName = ackSpan.getParentLevel() + "." + ackSpan.getLevelId();
}
// appending suffix
columnName += "-ACK";
put.addColumn(Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME), Bytes.toBytes(columnName), ackSpan.getData());
puts.add(put);
}
// save
HBaseUtil.batchSavePuts(connection, Config.HBaseConfig.TABLE_NAME, puts);
}
@Override
public int getType() {
return 2;
}
}

9
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/IProcessor.java Normal file
View File

@ -0,0 +1,9 @@
package com.ai.cloud.skywalking.reciever.processor;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import java.util.List;
public interface IProcessor {
void process(List<AbstractDataSerializable> serializedObjects);
}

26
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/ProcessorFactory.java Normal file
View File

@ -0,0 +1,26 @@
package com.ai.cloud.skywalking.reciever.processor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.HashMap;
import java.util.Map;
import java.util.ServiceLoader;
public class ProcessorFactory {
private static Logger logger = LogManager.getLogger(ProcessorFactory.class);
private static Map<Integer, AbstractSpanProcessor> type_processor_mapping = new HashMap<Integer, AbstractSpanProcessor>();
static {
ServiceLoader<AbstractSpanProcessor> processors = ServiceLoader.load(AbstractSpanProcessor.class);
for (AbstractSpanProcessor processor : processors) {
logger.info("Init protocol type and processor mapping : {} --> {}.", processor.getType(), processor.getClass().getName());
type_processor_mapping.put(processor.getType(), processor);
}
}
public static AbstractSpanProcessor chooseProcessor(int dataType) {
return type_processor_mapping.get(dataType);
}
}

51
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/RequestSpanProcessor.java Normal file
View File

@ -0,0 +1,51 @@
package com.ai.cloud.skywalking.reciever.processor;
import com.ai.cloud.skywalking.protocol.RequestSpan;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.util.HBaseUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import java.util.ArrayList;
import java.util.List;
import static com.ai.cloud.skywalking.reciever.util.SpanUtil.getTSBySpanTraceId;
public class RequestSpanProcessor extends AbstractSpanProcessor {
@Override
public void doAlarm(List<AbstractDataSerializable> serializedObjects) {
//DO Nothing
}
@Override
public void doSaveHBase(Connection connection, List<AbstractDataSerializable> serializedObjects) {
List<Put> puts = new ArrayList<Put>();
// convert to put
String columnName;
for (AbstractDataSerializable serializedObject : serializedObjects) {
RequestSpan requestSpan = (RequestSpan) serializedObject;
Put put = new Put(Bytes.toBytes(requestSpan.getTraceId()), getTSBySpanTraceId(requestSpan.getTraceId()));
if (StringUtils.isEmpty(requestSpan.getParentLevel().trim())) {
columnName = requestSpan.getLevelId() + "";
} else {
columnName = requestSpan.getParentLevel() + "." + requestSpan.getLevelId();
}
put.addColumn(Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME), Bytes.toBytes(columnName), requestSpan.getData());
puts.add(put);
}
// save
HBaseUtil.batchSavePuts(connection, Config.HBaseConfig.TABLE_NAME, puts);
}
@Override
public int getType() {
return 1;
}
}

7
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/exception/HBaseInitFailedException.java Normal file
View File

@ -0,0 +1,7 @@
package com.ai.cloud.skywalking.reciever.processor.exception;
public class HBaseInitFailedException extends RuntimeException {
public HBaseInitFailedException(String message, Exception e) {
super(message, e);
}
}

10
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/processor/exception/SaveToHBaseFailedException.java Normal file
View File

@ -0,0 +1,10 @@
package com.ai.cloud.skywalking.reciever.processor.exception;
/**
* Created by xin on 16-7-6.
*/
public class SaveToHBaseFailedException extends RuntimeException {
public SaveToHBaseFailedException(Exception e) {
super(e);
}
}

44
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/Chain.java
View File

@ -1,44 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage;
import com.ai.cloud.skywalking.protocol.Span;
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.util.List;
public class Chain {
private static Logger logger = LogManager
.getLogger(Chain.class);
private List<IStorageChain> chains;
private int index = 0;
public Chain(List<IStorageChain> chains) {
this.chains = chains;
}
public void doChain(List<Span> spans) {
if (index < chains.size()) {
while (true) {
try {
chains.get(index++).doChain(spans, this);
break;
} catch (Throwable e) {
logger.error("do chain at index[" + (index - 1) + "] failure.", e);
ServerHealthCollector.getCurrentHeathReading("storage-chain").updateData(ServerHeathReading.ERROR,
"do chain at index[" + (index - 1) + "] failure. spans list hash code:" + spans.hashCode() + ",Cause:" + e.getMessage());
// 如果Chain出现任何异常，将重做Chain,保证数据不丢失
index--;
}
}
}
}
void addChain(IStorageChain chain) {
chains.add(chain);
}
}

13
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/ChainException.java
View File

@ -1,13 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage;
public class ChainException extends RuntimeException {
private static final long serialVersionUID = -3134195788063272909L;
public ChainException(Throwable cause) {
super(cause);
}
public ChainException(String message, Throwable cause) {
super(message, cause);
}
}

10
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/IStorageChain.java
View File

@ -1,10 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage;
import com.ai.cloud.skywalking.protocol.Span;
import java.util.List;
public interface IStorageChain {
void doChain(List<Span> spans, Chain chain);
}

62
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/StorageChainController.java
View File

@ -1,62 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage;
import static com.ai.cloud.skywalking.reciever.conf.Config.StorageChain.STORAGE_TYPE;
import java.util.ArrayList;
import java.util.List;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.cloud.skywalking.reciever.conf.Constants;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
import com.ai.cloud.skywalking.reciever.storage.chain.AlarmChain;
import com.ai.cloud.skywalking.reciever.storage.chain.SaveToHBaseChain;
public class StorageChainController {
private static Logger logger = LogManager
.getLogger(StorageChainController.class);
private static List<IStorageChain> chainArray = new ArrayList<IStorageChain>();
static {
if (STORAGE_TYPE.equalsIgnoreCase("hbase")) {
chainArray.add(new AlarmChain());
chainArray.add(new SaveToHBaseChain());
} else {
throw new RuntimeException("illegal storage type.");
}
}
public static void doStorage(String buriedPointDatas) {
String[] buriedPointData = buriedPointDatas.split(Constants.DATA_SPILT);
if (buriedPointData == null || buriedPointData.length == 0) {
return;
}
List<Span> spans = new ArrayList<Span>();
for (String buriedPoint : buriedPointData) {
try {
if (buriedPoint == null || buriedPoint.trim().length() == 0) {
continue;
}
spans.add(new Span(buriedPoint));
} catch (Throwable e) {
logger.error("ready to save buriedPoint error, choose to ignore. data="
+ buriedPoint, e);
ServerHealthCollector.getCurrentHeathReading("StorageChainController").updateData(ServerHeathReading.ERROR,
"ready to save buriedPoint error, choose to ignore. data=" + buriedPoint);
}
}
try {
Chain chain = new Chain(chainArray);
chain.doChain(spans);
} catch (Throwable e) {
logger.error("Failed to storage chain.", e);
ServerHealthCollector.getCurrentHeathReading("StorageChainController").updateData(ServerHeathReading.ERROR,
"Failed to storage chain.Cause:" + e.getMessage());
}
}
}

45
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/AlarmChain.java
View File

@ -1,45 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage.chain;
import java.util.ArrayList;
import java.util.List;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.storage.Chain;
import com.ai.cloud.skywalking.reciever.storage.IStorageChain;
import com.ai.cloud.skywalking.reciever.storage.chain.alarm.ExceptionChecker;
import com.ai.cloud.skywalking.reciever.storage.chain.alarm.ExecuteTimeChecker;
import com.ai.cloud.skywalking.reciever.storage.chain.alarm.ISpanChecker;
import static com.ai.cloud.skywalking.reciever.conf.Config.Alarm.Checker.*;
public class AlarmChain implements IStorageChain {
private static Logger logger = LogManager.getLogger(AlarmChain.class);
private List<ISpanChecker> checkList = new ArrayList<ISpanChecker>();
public AlarmChain() {
if (TURN_ON_EXCEPTION_CHECKER)
checkList.add(new ExceptionChecker());
if (TURN_ON_EXECUTE_TIME_CHECKER)
checkList.add(new ExecuteTimeChecker());
}
@Override
public void doChain(List<Span> spans, Chain chain) {
if (Config.Alarm.ALARM_OFF_FLAG) {
return;
}
for (Span span : spans) {
for (ISpanChecker checker : checkList) {
checker.check(span);
}
}
chain.doChain(spans);
}
}

162
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/SaveToHBaseChain.java
View File

@ -1,162 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage.chain;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
import com.ai.cloud.skywalking.reciever.storage.Chain;
import com.ai.cloud.skywalking.reciever.storage.ChainException;
import com.ai.cloud.skywalking.reciever.storage.IStorageChain;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.mortbay.log.Log;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
public class SaveToHBaseChain implements IStorageChain {
private static Logger logger = LogManager.getLogger(SaveToHBaseChain.class);
private static Configuration configuration = null;
private static Connection connection;
@Override
public void doChain(List<Span> spans, Chain chain) {
if (connection == null || connection.isClosed()) {
initHBaseClient();
}
bulkInsertBuriedPointData(spans);
chain.doChain(spans);
}
private synchronized static void initHBaseClient() throws ChainException {
if (configuration == null) {
configuration = HBaseConfiguration.create();
if (Config.HBaseConfig.ZK_HOSTNAME == null
|| "".equals(Config.HBaseConfig.ZK_HOSTNAME)) {
logger.error("Miss HBase ZK quorum Configuration",
new IllegalArgumentException(
"Miss HBase ZK quorum Configuration"));
System.exit(-1);
}
configuration.set("hbase.zookeeper.quorum",
Config.HBaseConfig.ZK_HOSTNAME);
configuration.set("hbase.zookeeper.property.clientPort",
Config.HBaseConfig.CLIENT_PORT);
}
try {
connection = ConnectionFactory.createConnection(configuration);
} catch (IOException e) {
ServerHealthCollector.getCurrentHeathReading("hbase").updateData(
ServerHeathReading.ERROR, "connect to hbase failure.");
throw new ChainException("initHBaseClient failure", e);
}
}
static {
try {
initHBaseClient();
Admin admin = connection.getAdmin();
if (!admin.isTableAvailable(TableName
.valueOf(Config.HBaseConfig.TABLE_NAME))) {
HTableDescriptor tableDesc = new HTableDescriptor(
TableName.valueOf(Config.HBaseConfig.TABLE_NAME));
tableDesc.addFamily(new HColumnDescriptor(
Config.HBaseConfig.FAMILY_COLUMN_NAME));
admin.createTable(tableDesc);
logger.info("Create table [{}] ok!",
Config.HBaseConfig.TABLE_NAME);
}
} catch (IOException e) {
logger.error("Create table[{}] failed",
Config.HBaseConfig.TABLE_NAME, e);
}
}
private static void insert(String tableName, Put put) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
table.put(put);
} catch (IOException e) {
ServerHealthCollector.getCurrentHeathReading("hbase").updateData(
ServerHeathReading.ERROR,
"save RowKey[" + put.getId() + "] failure.");
throw new ChainException("Insert the data error.RowKey:["
+ put.getId() + "]", e);
}
}
private static void bulkInsertBuriedPointData(List<Span> spans) {
if (spans == null || spans.size() <= 0)
return;
List<Put> puts = new ArrayList<Put>();
Put put;
String columnName;
for (Span span : spans) {
put = new Put(Bytes.toBytes(span.getTraceId()),
getTSBySpanTraceId(span));
try {
if (StringUtils.isEmpty(span.getParentLevel().trim())) {
columnName = span.getLevelId() + "";
put.addColumn(
Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME),
Bytes.toBytes(columnName),
Bytes.toBytes(span.getOriginData()));
} else {
columnName = span.getParentLevel() + "." + span.getLevelId();
put.addColumn(
Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME),
Bytes.toBytes(columnName),
Bytes.toBytes(span.getOriginData()));
}
puts.add(put);
}catch (Throwable e){
logger.error("Failed to save to Span[{}] to hbase", span.toString());
throw e;
}
}
bulkInsertBuriedPointData(Config.HBaseConfig.TABLE_NAME, puts);
ServerHealthCollector.getCurrentHeathReading("hbase").updateData(
ServerHeathReading.INFO,
"save " + spans.size() + " BuriedPointEntries.");
}
private static long getTSBySpanTraceId(Span span) {
try {
return Long.parseLong(span.getTraceId().split("\\.")[2]);
} catch (Throwable t) {
Log.warn("can't get timestamp from trace id:{}, going to use current timestamp.", span.getTraceId(), t);
return System.currentTimeMillis();
}
}
private static void bulkInsertBuriedPointData(String tableName,
List<Put> data) {
Object[] resultArrays = new Object[data.size()];
try {
Table table = connection.getTable(TableName.valueOf(tableName));
table.batch(data, resultArrays);
int index = 0;
for (Object result : resultArrays) {
if (result != null) {
insert(tableName, data.get(index));
}
index++;
}
} catch (IOException e) {
throw new ChainException(e);
} catch (InterruptedException e) {
throw new ChainException(e);
}
}
}

30
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/alarm/AbstractSpanChecker.java
View File

@ -1,30 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage.chain.alarm;
import static com.ai.cloud.skywalking.reciever.conf.Config.Alarm.ALARM_EXPIRE_SECONDS;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import redis.clients.jedis.Jedis;
import com.ai.cloud.skywalking.reciever.util.RedisConnector;
public abstract class AbstractSpanChecker implements ISpanChecker {
private static Logger logger = LogManager.getLogger(AbstractSpanChecker.class);
protected void saveAlarmMessage(String key, String traceId, String alarmMsg) {
Jedis jedis = null;
try {
jedis = RedisConnector.getJedis();
jedis.hsetnx(key, traceId, alarmMsg);
jedis.expire(key, ALARM_EXPIRE_SECONDS);
} catch (Exception e) {
RedisConnector.reportJedisFailure();
logger.error("Failed to set data.", e);
} finally {
if (jedis != null) {
jedis.close();
}
}
}
}

31
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/alarm/ExceptionChecker.java
View File

@ -1,31 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage.chain.alarm;
import static com.ai.cloud.skywalking.reciever.conf.Config.Alarm.ALARM_EXCEPTION_STACK_LENGTH;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.ai.cloud.skywalking.protocol.Span;
public class ExceptionChecker extends AbstractSpanChecker {
private static Logger logger = LogManager.getLogger(ExceptionChecker.class);
@Override
public void check(Span span) {
if (span.getStatusCode() != 1)
return;
String exceptionStack = span.getExceptionStack();
if (exceptionStack == null) {
exceptionStack = "";
} else if (exceptionStack.length() > ALARM_EXCEPTION_STACK_LENGTH) {
exceptionStack = exceptionStack.substring(0, ALARM_EXCEPTION_STACK_LENGTH);
}
saveAlarmMessage(generateAlarmKey(span), span.getTraceId(), exceptionStack);
}
private String generateAlarmKey(Span span) {
return span.getUserId() + "-" + span.getApplicationId() + "-"
+ (System.currentTimeMillis() / (10000 * 6));
}
}

36
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/alarm/ExecuteTimeChecker.java
View File

@ -1,36 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage.chain.alarm;
import com.ai.cloud.skywalking.protocol.Span;
public class ExecuteTimeChecker extends AbstractSpanChecker {
@Override
public void check(Span span) {
long cost = span.getCost();
if (cost > 500 && cost < 3000) {
/**
* Issue #43 <br/>
* 单埋点调用时间超过500ms的进行预警
*/
saveAlarmMessage(generateWarningAlarmKey(span), span.getTraceId(), span.getViewPointId() + " cost " + cost + " ms.");
}
if (cost >= 3000) {
/**
* Issue #43 <br/>
* 单埋点调用时间超过3S的进行告警
*/
saveAlarmMessage(generatePossibleErrorAlarmKey(span), span.getTraceId(), span.getViewPointId() + " cost " + cost + " ms.");
}
}
private String generateWarningAlarmKey(Span span) {
return span.getUserId() + "-" + span.getApplicationId() + "-"
+ (System.currentTimeMillis() / (10000 * 6)) + "-ExecuteTime-Warning";
}
private String generatePossibleErrorAlarmKey(Span span) {
return span.getUserId() + "-" + span.getApplicationId() + "-"
+ (System.currentTimeMillis() / (10000 * 6)) + "-ExecuteTime-PossibleError";
}
}

7
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/alarm/ISpanChecker.java
View File

@ -1,7 +0,0 @@
package com.ai.cloud.skywalking.reciever.storage.chain.alarm;
import com.ai.cloud.skywalking.protocol.Span;
public interface ISpanChecker {
void check(Span span);
}

26
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/HBaseUtil.java Normal file
View File

@ -0,0 +1,26 @@
package com.ai.cloud.skywalking.reciever.util;
import com.ai.cloud.skywalking.reciever.processor.exception.SaveToHBaseFailedException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import java.io.IOException;
import java.util.List;
public class HBaseUtil {
public static void batchSavePuts(Connection connection, String tableName, List<Put> puts) {
Object[] resultArrays = new Object[puts.size()];
try {
Table table = connection.getTable(TableName.valueOf(tableName));
table.batch(puts, resultArrays);
// ignore failed data
} catch (IOException e) {
throw new SaveToHBaseFailedException(e);
} catch (InterruptedException e) {
throw new SaveToHBaseFailedException(e);
}
}
}

18
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/SpanUtil.java Normal file
View File

@ -0,0 +1,18 @@
package com.ai.cloud.skywalking.reciever.util;
import org.mortbay.log.Log;
/**
* Created by xin on 16-7-6.
*/
public class SpanUtil {
public static long getTSBySpanTraceId(String traceId) {
try {
return Long.parseLong(traceId.split("\\.")[2]);
} catch (Throwable t) {
Log.warn("can't get timestamp from trace id:{}, going to use current timestamp.", traceId, t);
return System.currentTimeMillis();
}
}
}

2
skywalking-server/src/main/resources/META-INF/services/com.ai.cloud.skywalking.reciever.processor.AbstractSpanProcessor Normal file
View File

@ -0,0 +1,2 @@
com.ai.cloud.skywalking.reciever.processor.RequestSpanProcessor
com.ai.cloud.skywalking.reciever.processor.AckSpanProcessor