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

1. 重构代码 2. 修复序列化问题

This commit is contained in:
ascrutae 2016-08-04 13:30:35 +08:00
parent d655c23c36
commit b6d262c2fa
33 changed files with 1107 additions and 467 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
samples/skywalking-auth/src/main/resources/sky-walking.auth
View File

@ -31,7 +31,7 @@ sender.max_send_length=20000
sender.retry_get_sender_wait_interval=2000
#最大消费线程数
consumer.max_consumer=0
consumer.max_consumer=1
#消费者最大等待时间
consumer.max_wait_time=5
#发送失败等待时间

4
samples/skywalking-example/example-dubbo/dubbo-impl/src/main/resources/provider/dubbo-provider.xml
View File

@ -8,8 +8,8 @@
http://code.alibabatech.com/schema/dubbo/dubbo.xsd">
<dubbo:application name="skywalking-sample-dubbo-impl" />
<dubbo:registry address="zookeeper://127.0.0.1:2181" />
<dubbo:registry address="zookeeper://127.0.0.1:2181" timeout="50000" />
<dubbo:protocol name="dubbo" port="20880" />
<dubbo:annotation package="com.ai.cloud.skywalking.sample.dubbo"/>
</beans>
</beans>

4
samples/skywalking-example/example-web/src/main/resources/consumer/dubbo-consumer.xml
View File

@ -7,6 +7,6 @@
http://code.alibabatech.com/schema/dubbo
http://code.alibabatech.com/schema/dubbo/dubbo.xsd">
<dubbo:application name="skywalking-sample-dubbo-consumer"/>
<dubbo:registry protocol="zookeeper" address="127.0.0.1:2181"/>
<dubbo:registry protocol="zookeeper" address="127.0.0.1:2181" timeout="50000"/>
<dubbo:reference id="sampleDubboInterface" interface="com.ai.cloud.skywalking.sample.dubbo.interfaces.SampleDubboInterface"/>
</beans>
</beans>

45
samples/skywalking-example/example-web/src/main/resources/sky-walking.auth
View File

@ -1,45 +0,0 @@
#skyWalking用户ID
skywalking.user_id=123
#skyWalking应用编码
skywalking.application_code=skywalking-sample-dubbo
#skywalking auth的环境变量名字
skywalking.auth_system_env_name=SKYWALKING_RUN
#skywalking数据编码
skywalking.charset=UTF-8
skywalking.auth_override=true
#是否打印数据
buriedpoint.printf=true
#埋点异常的最大长度
buriedpoint.max_exception_stack_length=4000
#业务字段的最大长度
buriedpoint.businesskey_max_length=300
#过滤异常
buriedpoint.exclusive_exceptions=java.lang.RuntimeException
#最大发送者的连接数阀比例
sender.connect_percent=100
#发送服务端配置
sender.servers_addr=127.0.0.1:34000
#最大发送的副本数量
sender.max_copy_num=2
#发送的最大长度
sender.max_send_length=20000
#当没有Sender时,尝试获取sender的等待周期
sender.retry_get_sender_wait_interval=2000
#最大消费线程数
consumer.max_consumer=0
#消费者最大等待时间
consumer.max_wait_time=5
#发送失败等待时间
consumer.consumer_fail_retry_wait_interval=50
#每个Buffer的最大个数
buffer.buffer_max_size=18000
#Buffer池的最大长度
buffer.pool_size=5
#发送检查线程检查周期
senderchecker.check_polling_time=200

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

@ -33,14 +33,6 @@ public class CurrentThreadSpanStack {
return nodes.get().pop();
}
public static void invalidatePresentSpans() {
if (nodes.get() == null) {
nodes.set(new SpanNodeStack());
}
nodes.get().invalidatePresentSpans();
}
static class SpanNodeStack {
/**
* 单JVM的单线程,埋点数量一般不会超过20.
@ -84,11 +76,6 @@ public class CurrentThreadSpanStack {
spans.add(spans.size(), spanNode);
}
public void invalidatePresentSpans() {
for (SpanNode spanNode : spans) {
spanNode.getData().setValidate(true);
}
}
}
static class SpanNode {

5
skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/plugin/interceptor/enhance/ClassEnhancePluginDefine.java
View File

@ -1,5 +1,6 @@
package com.ai.cloud.skywalking.plugin.interceptor.enhance;
import static net.bytebuddy.jar.asm.Opcodes.ACC_PRIVATE;
import static net.bytebuddy.matcher.ElementMatchers.any;
import static net.bytebuddy.matcher.ElementMatchers.not;
@ -20,6 +21,8 @@ import com.ai.cloud.skywalking.plugin.interceptor.EnhanceException;
import com.ai.cloud.skywalking.plugin.interceptor.EnhancedClassInstanceContext;
import com.ai.cloud.skywalking.plugin.interceptor.MethodMatcher;
import java.lang.reflect.Modifier;
public abstract class ClassEnhancePluginDefine extends AbstractClassEnhancePluginDefine {
private static Logger logger = LogManager
.getLogger(ClassEnhancePluginDefine.class);
@ -61,7 +64,7 @@ public abstract class ClassEnhancePluginDefine extends AbstractClassEnhancePlugi
newClassBuilder = newClassBuilder
.defineField(contextAttrName,
EnhancedClassInstanceContext.class)
EnhancedClassInstanceContext.class, ACC_PRIVATE)
.constructor(any())
.intercept(
SuperMethodCall.INSTANCE.andThen(MethodDelegation.to(

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

@ -21,15 +21,16 @@
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>2.6.1</version>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>

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

File diff suppressed because it is too large Load Diff

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

@ -213,11 +213,20 @@ public class RequestSpan extends AbstractDataSerializable {
@Override
public byte[] getData() {
return TraceProtocol.RequestSpan.newBuilder().setTraceId(traceId).setParentLevel(parentLevel)
.setLevelId(levelId).setViewPointId(viewPointId).setStartDate(startDate)
.setSpanType(spanType.getValue()).setSpanTypeDesc(spanTypeDesc).setBussinessKey(businessKey)
.setCallType(callType).setApplicationId(applicationId).setUserId(userId).setBussinessKey(businessKey)
.setAgentId(agentId).build().toByteArray();
TraceProtocol.RequestSpan.Builder builder =
TraceProtocol.RequestSpan.newBuilder().setTraceId(traceId).setParentLevel(parentLevel)
.setLevelId(levelId).setViewPointId(viewPointId).setStartDate(startDate)
.setSpanType(spanType.getValue()).setSpanTypeDesc(spanTypeDesc);
if (businessKey != null && businessKey.length() > 0) {
builder.setBussinessKey(businessKey);
}
if (parameters != null && parameters.size() > 0) {
builder.getParametersMap().putAll(parameters);
}
return builder.setCallType(callType).setApplicationId(applicationId).setUserId(userId).setAgentId(agentId)
.build().toByteArray();
}
@Override

2
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializedFactory.java
View File

@ -22,7 +22,7 @@ public class SerializedFactory {
}
}
public static AbstractDataSerializable unSerialize(byte[] bytes) throws ConvertFailedException {
public static AbstractDataSerializable deserialize(byte[] bytes) throws ConvertFailedException {
try {
AbstractDataSerializable abstractDataSerializable = serializableMap.get(IntegerAssist.bytesToInt(bytes, 0));
if (abstractDataSerializable != null) {

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

@ -47,12 +47,6 @@ public class Span {
*/
protected String exceptionStack = "";
/**
* 节点的状态<br/>
* 不参与序列化
*/
protected boolean isValidate = true;
/**
* 节点调用过程中的业务字段<br/>
* 如：业务系统设置的订单号，SQL语句等
@ -82,6 +76,7 @@ public class Span {
this.traceId = traceId;
this.applicationId = applicationId;
this.userId = userId;
this.parentLevel = "";
}
public Span(String traceId, String parentLevel, int levelId, String applicationId, String userId) {
@ -120,10 +115,6 @@ public class Span {
this.startDate = startDate;
}
public boolean isValidate() {
return isValidate;
}
public byte getStatusCode() {
return statusCode;
}
@ -144,17 +135,6 @@ public class Span {
this.parameters = parameters;
}
public void setValidate(boolean validate) {
isValidate = validate;
}
public boolean isRPCClientSpan() {
if (spanType == SpanType.RPC_CLIENT) {
return true;
}
return false;
}
public void setSpanType(SpanType spanType) {
this.spanType = spanType;
}
@ -228,10 +208,6 @@ public class Span {
this.viewPointId = viewPointId;
}
public void appendParameter(String key, String value) {
this.parameters.put(key, value);
}
public void setInvokeResult(String result){
this.parameters.put(INVOKE_RESULT_PARAMETER_KEY, result);
}

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

@ -9,58 +9,21 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class TransportPackager {
import static com.ai.cloud.skywalking.protocol.util.ByteDataUtil.generateChecksum;
public class TransportPackager {
public static byte[] pack(List<ISerializable> beSendingData) {
// 对协议格式进行修改
// | check sum(4 byte) | data
byte[] dataText = packSerializableObjects(beSendingData);
byte[] dataPackage = packCheckSum(dataText);
byte[] data = serializeObjects(beSendingData);
byte[] dataPackage = appendCheckSum(data);
return dataPackage;
}
public static List<AbstractDataSerializable> unpackSerializableObjects(byte[] dataPackage) {
List<AbstractDataSerializable> serializeData = new ArrayList<AbstractDataSerializable>();
int currentLength = 0;
while (true) {
//读取长度
int dataLength = IntegerAssist.bytesToInt(dataPackage, currentLength);
// 反序列化
byte[] data = new byte[dataLength];
System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength);
try {
AbstractDataSerializable abstractDataSerializable = SerializedFactory.unSerialize(data);
serializeData.add(abstractDataSerializable);
} catch (ConvertFailedException e) {
e.printStackTrace();
}
currentLength += 4 + dataLength;
if (currentLength >= dataPackage.length) {
break;
}
}
return serializeData;
}
/**
* 生成校验和参数
*
* @param data
* @return
*/
private static byte[] generateChecksum(byte[] data, int offset) {
int result = data[offset];
for (int i = offset + 1; i < data.length; i++) {
result ^= data[i];
}
return IntegerAssist.intToBytes(result);
}
private static byte[] packCheckSum(byte[] dataText) {
private static byte[] appendCheckSum(byte[] dataText) {
byte[] dataPackage = new byte[dataText.length + 4];
byte[] checkSum = generateChecksum(dataText, 0);
System.arraycopy(checkSum, 0, dataPackage, 0, 4);
@ -68,19 +31,20 @@ public class TransportPackager {
return dataPackage;
}
private static byte[] packSerializableObjects(List<ISerializable> beSendingData) {
byte[] dataText = null;
private static byte[] serializeObjects(List<ISerializable> beSendingData) {
byte[] data = null;
int currentIndex = 0;
for (ISerializable sendingData : beSendingData) {
byte[] dataElementText = packSerializableObject(sendingData);
dataText = appendingDataBytes(dataText, currentIndex, dataElementText);
currentIndex += dataElementText.length;
byte[] elementData = serialize(sendingData);
data = appendData(data, currentIndex, elementData);
currentIndex += elementData.length;
}
return dataText;
return data;
}
private static byte[] appendingDataBytes(byte[] dataText, int currentIndex, byte[] dataElementText) {
private static byte[] appendData(byte[] dataText, int currentIndex, byte[] dataElementText) {
if (dataText == null) {
dataText = new byte[dataElementText.length];
} else {
@ -91,9 +55,10 @@ public class TransportPackager {
}
public static byte[] packSerializableObject(ISerializable serializable) {
public static byte[] serialize(ISerializable serializable) {
byte[] serializableBytes = serializable.convert2Bytes();
byte[] dataText = Arrays.copyOf(serializableBytes, serializableBytes.length + 4);
byte[] dataText = new byte[serializableBytes.length + 4];
System.arraycopy(serializableBytes, 0, dataText, 4, serializableBytes.length);
byte[] length = IntegerAssist.intToBytes(serializableBytes.length);
System.arraycopy(length, 0, dataText, 0, 4);
return dataText;

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

@ -23,7 +23,7 @@ public enum SpanType {
return LOCAL;
case 2:
return RPC_CLIENT;
case 3:
case 4:
return RPC_SERVER;
default:
throw new SpanTypeCannotConvertException(spanTypeValue + "");

28
skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/util/ByteDataUtil.java Normal file
View File

@ -0,0 +1,28 @@
package com.ai.cloud.skywalking.protocol.util;
import java.util.Arrays;
public class ByteDataUtil {
public static byte[] unpackCheckSum(byte[] msg) {
return Arrays.copyOfRange(msg, 4, msg.length);
}
public static boolean validateCheckSum(byte[] dataPackage) {
byte[] checkSum = generateChecksum(dataPackage, 4);
byte[] originCheckSum = new byte[4];
System.arraycopy(dataPackage, 0, originCheckSum, 0, 4);
return Arrays.equals(checkSum, originCheckSum);
}
public static byte[] generateChecksum(byte[] data, int offset) {
int result = data[offset];
for (int i = offset + 1; i < data.length; i++) {
result ^= data[i];
}
return IntegerAssist.intToBytes(result);
}
}

1
skywalking-collector/skywalking-protocol/src/main/proto/TraceProtocol.proto
View File

@ -25,4 +25,5 @@ message RequestSpan {
required string userId = 10;
optional string bussinessKey = 11;
required string agentId = 12;
map<string,string> parameters = 13;
}

5
skywalking-collector/skywalking-sdk-plugin/dubbo-plugin/pom.xml
View File

@ -97,20 +97,21 @@
<scope>compile</scope>
</dependency>
-->
<!--
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>dubbox</artifactId>
<version>2.8.4</version>
<scope>provided</scope>
</dependency>
<!--
-->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>dubbo</artifactId>
<version>2.5.3</version>
<scope>provided</scope>
</dependency>
-->
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>

26
skywalking-collector/skywalking-sdk-plugin/dubbo-plugin/src/main/java/com/ai/cloud/skywalking/plugin/dubbo/MonitorFilterInterceptor.java
View File

@ -33,9 +33,7 @@ public class MonitorFilterInterceptor implements InstanceMethodsAroundIntercepto
boolean isConsumer = rpcContext.isConsumerSide();
context.set("isConsumer", isConsumer);
if (isConsumer) {
RPCClientInvokeMonitor rpcClientInvokeMonitor = new RPCClientInvokeMonitor();
context.set("rpcClientInvokeMonitor", rpcClientInvokeMonitor);
ContextData contextData = rpcClientInvokeMonitor.beforeInvoke(createIdentification(invoker, invocation));
ContextData contextData = new RPCClientInvokeMonitor().beforeInvoke(createIdentification(invoker, invocation));
String contextDataStr = contextData.toString();
//追加参数
@ -60,8 +58,6 @@ public class MonitorFilterInterceptor implements InstanceMethodsAroundIntercepto
}
} else {
// 读取参数
RPCServerInvokeMonitor rpcServerInvokeMonitor = new RPCServerInvokeMonitor();
context.set("rpcServerInvokeMonitor", rpcServerInvokeMonitor);
String contextDataStr;
if (!BugFixAcitve.isActive) {
@ -75,7 +71,7 @@ public class MonitorFilterInterceptor implements InstanceMethodsAroundIntercepto
contextData = new ContextData(contextDataStr);
}
rpcServerInvokeMonitor.beforeInvoke(contextData, createIdentification(invoker, invocation));
new RPCServerInvokeMonitor().beforeInvoke(contextData, createIdentification(invoker, invocation));
}
}
@ -88,6 +84,12 @@ public class MonitorFilterInterceptor implements InstanceMethodsAroundIntercepto
dealException(result.getException(), context);
}
if (isConsumer(context)){
new RPCClientInvokeMonitor().afterInvoke();
}else{
new RPCServerInvokeMonitor().afterInvoke();
}
return ret;
}
@ -97,12 +99,16 @@ public class MonitorFilterInterceptor implements InstanceMethodsAroundIntercepto
dealException(t, context);
}
private boolean isConsumer(EnhancedClassInstanceContext context){
return (boolean) context.get("isConsumer");
}
private void dealException(Throwable t, EnhancedClassInstanceContext context) {
boolean isConsumer = (boolean) context.get("isConsumer");
if (isConsumer) {
((RPCClientInvokeMonitor) context.get("rpcClientInvokeMonitor")).occurException(t);
if (isConsumer(context)) {
new RPCClientInvokeMonitor().occurException(t);
} else {
((RPCServerInvokeMonitor) context.get("rpcServerInvokeMonitor")).occurException(t);
new RPCServerInvokeMonitor().occurException(t);
}
}

17
skywalking-collector/skywalking-sdk-plugin/jdbc-plugin/src/main/java/com/ai/cloud/skywalking/plugin/jdbc/define/AbstractDatabasePluginDefine.java Normal file
View File

@ -0,0 +1,17 @@
package com.ai.cloud.skywalking.plugin.jdbc.define;
import com.ai.cloud.skywalking.plugin.interceptor.MethodMatcher;
import com.ai.cloud.skywalking.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
import com.ai.cloud.skywalking.plugin.interceptor.matcher.SimpleMethodMatcher;
public abstract class AbstractDatabasePluginDefine extends ClassInstanceMethodsEnhancePluginDefine {
@Override
protected MethodMatcher[] getInstanceMethodsMatchers() {
return new MethodMatcher[]{new SimpleMethodMatcher("connect")};
}
@Override
protected String getInstanceMethodsInterceptor() {
return "com.ai.cloud.skywalking.plugin.jdbc.define.DatabasePluginInterceptor";
}
}

36
skywalking-collector/skywalking-sdk-plugin/jdbc-plugin/src/main/java/com/ai/cloud/skywalking/plugin/jdbc/define/DatabasePluginInterceptor.java Normal file
View File

@ -0,0 +1,36 @@
package com.ai.cloud.skywalking.plugin.jdbc.define;
import com.ai.cloud.skywalking.plugin.interceptor.EnhancedClassInstanceContext;
import com.ai.cloud.skywalking.plugin.interceptor.enhance.ConstructorInvokeContext;
import com.ai.cloud.skywalking.plugin.interceptor.enhance.InstanceMethodInvokeContext;
import com.ai.cloud.skywalking.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import com.ai.cloud.skywalking.plugin.interceptor.enhance.MethodInterceptResult;
import com.ai.cloud.skywalking.plugin.jdbc.SWConnection;
import java.sql.Connection;
import java.util.Properties;
public class DatabasePluginInterceptor implements InstanceMethodsAroundInterceptor {
@Override
public void onConstruct(EnhancedClassInstanceContext context, ConstructorInvokeContext interceptorContext) {
}
@Override
public void beforeMethod(EnhancedClassInstanceContext context, InstanceMethodInvokeContext interceptorContext,
MethodInterceptResult result) {
}
@Override
public Object afterMethod(EnhancedClassInstanceContext context, InstanceMethodInvokeContext interceptorContext,
Object ret) {
return new SWConnection((String) interceptorContext.allArguments()[0],
(Properties) interceptorContext.allArguments()[1], (Connection) ret);
}
@Override
public void handleMethodException(Throwable t, EnhancedClassInstanceContext context,
InstanceMethodInvokeContext interceptorContext, Object ret) {
}
}

10
skywalking-collector/skywalking-sdk-plugin/jdbc-plugin/src/main/java/com/ai/cloud/skywalking/plugin/jdbc/plugin/H2DatabasePluginDefine.java Normal file
View File

@ -0,0 +1,10 @@
package com.ai.cloud.skywalking.plugin.jdbc.plugin;
import com.ai.cloud.skywalking.plugin.jdbc.define.AbstractDatabasePluginDefine;
public class H2DatabasePluginDefine extends AbstractDatabasePluginDefine {
@Override
protected String enhanceClassName() {
return "org.h2.Driver";
}
}

2
skywalking-collector/skywalking-sdk-plugin/jdbc-plugin/src/main/resources/skywalking-plugin.def
View File

@ -1 +1 @@
com.ai.cloud.skywalking.plugin.jdbc.JDBCPluginDefine
com.ai.cloud.skywalking.plugin.jdbc.plugin.H2DatabasePluginDefine

3
skywalking-collector/skywalking-sdk-plugin/tomcat-7.x-8.x-plugin/src/main/java/com/ai/cloud/skywalking/plugin/tomcat78x/TomcatPluginInterceptor.java
View File

@ -62,11 +62,12 @@ public class TomcatPluginInterceptor implements InstanceMethodsAroundInterceptor
Object[] args = interceptorContext.allArguments();
HttpServletResponse httpServletResponse = (HttpServletResponse) args[1];
httpServletResponse.addHeader(TRACE_ID_HEADER_NAME, Tracing.getTraceId());
new RPCServerInvokeMonitor().afterInvoke();
return ret;
}
@Override
public void handleMethodException(Throwable t, EnhancedClassInstanceContext context, InstanceMethodInvokeContext interceptorContext, Object ret) {
// DO Nothing
new RPCServerInvokeMonitor().occurException(t);
}
}

5
skywalking-server/pom.xml
View File

@ -56,6 +56,11 @@
<artifactId>gson</artifactId>
<version>2.2.2</version>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.0.0</version>
</dependency>
</dependencies>
<build>
<plugins>

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

@ -2,7 +2,6 @@ package com.ai.cloud.skywalking.reciever.buffer;
import com.ai.cloud.skywalking.protocol.BufferFileEOFProtocol;
import com.ai.cloud.skywalking.protocol.TransportPackager;
import com.ai.cloud.skywalking.reciever.model.BufferDataPackagerGenerator;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -29,8 +28,8 @@ class AppendEOFFlagThread extends Thread {
try {
logger.info("Add EOF flags to unprocessed data file[{}]", file.getName());
fileOutputStream = new FileOutputStream(new File(file.getParent(), file.getName()), true);
fileOutputStream.write(BufferDataPackagerGenerator
.pack(TransportPackager.packSerializableObject(new BufferFileEOFProtocol())));
fileOutputStream.write(BufferDataAssist
.appendLengthAndSplit(TransportPackager.serialize(new BufferFileEOFProtocol())));
} catch (IOException e) {
logger.info("Add EOF flags to the unprocessed data file failed.", e);
} finally {

28
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BufferDataPackagerGenerator.java → skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/BufferDataAssist.java
View File

@ -1,31 +1,23 @@
package com.ai.cloud.skywalking.reciever.model;
package com.ai.cloud.skywalking.reciever.buffer;
import com.ai.cloud.skywalking.protocol.SerializedFactory;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
public class BufferDataPackagerGenerator {
import java.util.ArrayList;
import java.util.List;
public class BufferDataAssist {
private static byte[] SPILT = new byte[] {127, 127, 127, 127};
private static byte[] EOF = null;
private BufferDataPackagerGenerator() {
private BufferDataAssist() {
//DO Nothing
}
public static byte[] generateEOFPackage() {
if (EOF != null) {
return EOF;
}
EOF = generatePackage("EOF".getBytes());
return EOF;
}
public static byte[] pack(byte[] msg) {
return generatePackage(msg);
}
private static byte[] generatePackage(byte[] msg) {
public static byte[] appendLengthAndSplit(byte[] msg) {
byte[] dataPackage = new byte[msg.length + 8];
// 前四位长度
System.arraycopy(IntegerAssist.intToBytes(msg.length), 0, dataPackage, 0, 4);

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

@ -4,7 +4,6 @@ import com.ai.cloud.skywalking.protocol.BufferFileEOFProtocol;
import com.ai.cloud.skywalking.protocol.TransportPackager;
import com.ai.cloud.skywalking.protocol.util.AtomicRangeInteger;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.model.BufferDataPackagerGenerator;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
import org.apache.logging.log4j.LogManager;
@ -44,7 +43,7 @@ public class DataBufferThread extends Thread {
}
try {
fileOutputStream.write(BufferDataPackagerGenerator.pack(data[i]));
fileOutputStream.write(BufferDataAssist.appendLengthAndSplit(data[i]));
length += data[i].length;
data[i] = null;
} catch (IOException e) {
@ -69,8 +68,8 @@ public class DataBufferThread extends Thread {
private void closeCurrentBufferFile(FileOutputStream fileOutputStream) {
try {
fileOutputStream.flush();
fileOutputStream.write(BufferDataPackagerGenerator
.pack(TransportPackager.packSerializableObject(new BufferFileEOFProtocol())));
fileOutputStream.write(BufferDataAssist
.appendLengthAndSplit(TransportPackager.serialize(new BufferFileEOFProtocol())));
} catch (IOException e) {
logger.error("Failed to write msg.", e);
} finally {

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

@ -58,10 +58,9 @@ public class DataBufferThreadContainer {
countDownLatch.await();
}
logger.info("Data buffer thread size {} begin to init ", Config.Server.
MAX_DEAL_DATA_THREAD_NUMBER);
logger.info("Data buffer thread size {} begin to init ", Config.Buffer.BUFFER_DEAL_THREAD_NUMBER);
for (int i = 0; i < Config.Server.MAX_DEAL_DATA_THREAD_NUMBER; i++) {
for (int i = 0; i < Config.Buffer.BUFFER_DEAL_THREAD_NUMBER; i++) {
DataBufferThread dataBufferThread = new DataBufferThread(i);
dataBufferThread.start();
buffers.add(dataBufferThread);

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

@ -6,8 +6,7 @@ public class Config {
public static class Server {
// 采集服务器的端口
public static int PORT = 34000;
// 最大数据处理线程数量
public static int MAX_DEAL_DATA_THREAD_NUMBER = 3;
// 异常数据的时间间隔
public static int FAILED_PACKAGE_WATCHING_TIME_WINDOW = 5 * 60;
// 时间间隔内最大异常数据次数
@ -31,6 +30,8 @@ public class Config {
public static long BUFFER_FILE_MAX_LENGTH = 30 * 1024 * 1024;
public static int BUFFER_DEAL_THREAD_NUMBER = 0;
}
@ -50,6 +51,9 @@ public class Config {
public static class Persistence {
// 最大数据处理线程数量
public static int MAX_DEAL_DATA_THREAD_NUMBER = 3;
// 切换文件，等待时间
public static long SWITCH_FILE_WAIT_TIME = 5000L;

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

@ -1,6 +1,5 @@
package com.ai.cloud.skywalking.reciever.handler;
import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
import com.ai.cloud.skywalking.reciever.buffer.DataBufferThreadContainer;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.util.RedisConnector;
@ -9,7 +8,9 @@ import io.netty.channel.SimpleChannelInboundHandler;
import redis.clients.jedis.Jedis;
import java.net.InetSocketAddress;
import java.util.Arrays;
import static com.ai.cloud.skywalking.protocol.util.ByteDataUtil.unpackCheckSum;
import static com.ai.cloud.skywalking.protocol.util.ByteDataUtil.validateCheckSum;
public class CollectionServerDataHandler extends SimpleChannelInboundHandler<byte[]> {
@ -27,26 +28,6 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler<byt
}
}
private byte[] unpackCheckSum(byte[] msg) {
return Arrays.copyOfRange(msg, 4, msg.length);
}
private boolean validateCheckSum(byte[] dataPackage) {
byte[] checkSum = generateChecksum(dataPackage, 4);
byte[] originCheckSum = new byte[4];
System.arraycopy(dataPackage, 0, originCheckSum, 0, 4);
return Arrays.equals(checkSum, originCheckSum);
}
private static byte[] generateChecksum(byte[] data, int offset) {
int result = data[offset];
for (int i = offset + 1; i < data.length; i++) {
result ^= data[i];
}
return IntegerAssist.intToBytes(result);
}
private void dealFailedPackage(ChannelHandlerContext ctx) {
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().localAddress();

51
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/BufferFileReader.java
View File

@ -1,7 +1,8 @@
package com.ai.cloud.skywalking.reciever.peresistent;
import com.ai.cloud.skywalking.protocol.TransportPackager;
import com.ai.cloud.skywalking.protocol.SerializedFactory;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.exception.ConvertFailedException;
import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector;
import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading;
@ -20,10 +21,10 @@ public class BufferFileReader {
private File bufferFile;
private FileInputStream bufferInputStream;
private int currentOffset;
private static final byte[] SPILT_BALE_ARRAY = new byte[] {127, 127, 127, 127};
private int remainderLength = 0;
private byte[] remainderByte = null;
private Logger logger = LogManager.getLogger(BufferFileReader.class);
private static final byte[] DATA_SPILT = new byte[] {127, 127, 127, 127};
private int remainderLength = 0;
private byte[] remainderByte = null;
private Logger logger = LogManager.getLogger(BufferFileReader.class);
private List<AbstractDataSerializable> serializables;
public BufferFileReader(File bufferFile, int currentOffset) {
@ -58,13 +59,12 @@ public class BufferFileReader {
int length = unpackLength();
byte[] dataContext = readByte(length);
// 转换对象
serializables = new ArrayList<>();
serializables = TransportPackager.unpackSerializableObjects(dataContext);
serializables = deserializableObjects(dataContext);
byte[] skip = new byte[4];
bufferInputStream.read(skip);
if (!Arrays.equals(SPILT_BALE_ARRAY, skip)) {
skipToNextBufferBale();
if (!Arrays.equals(DATA_SPILT, skip)) {
skipToNext();
}
MemoryRegister.instance().updateOffSet(bufferFile.getName(), currentOffset);
} catch (IOException e) {
@ -122,14 +122,14 @@ public class BufferFileReader {
return lengthByte;
}
public void skipToNextBufferBale() throws IOException {
public void skipToNext() throws IOException {
byte[] previousDataByte = new byte[4];
byte[] currentDataByte = new byte[4];
byte[] compactDataByte = new byte[8];
while (true) {
currentDataByte = readByte(4);
if (Arrays.equals(currentDataByte, SPILT_BALE_ARRAY)) {
if (Arrays.equals(currentDataByte, DATA_SPILT)) {
remainderLength = 0;
break;
}
@ -137,7 +137,7 @@ public class BufferFileReader {
System.arraycopy(previousDataByte, 0, compactDataByte, 0, 4);
System.arraycopy(currentDataByte, 0, compactDataByte, 4, 4);
int index = bytesIndexOf(compactDataByte, SPILT_BALE_ARRAY, 0, 8);
int index = bytesIndexOf(compactDataByte, DATA_SPILT, 0, 8);
if (index != -1) {
recodeRemainderByteAndLength(compactDataByte, index);
break;
@ -185,4 +185,31 @@ public class BufferFileReader {
bufferFile = null;
}
public static List<AbstractDataSerializable> deserializableObjects(byte[] dataPackage) {
List<AbstractDataSerializable> serializeData = new ArrayList<AbstractDataSerializable>();
int currentLength = 0;
while (true) {
//读取长度
int dataLength = IntegerAssist.bytesToInt(dataPackage, currentLength);
// 反序列化
byte[] data = new byte[dataLength];
System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength);
try {
AbstractDataSerializable abstractDataSerializable = SerializedFactory.deserialize(data);
serializeData.add(abstractDataSerializable);
} catch (ConvertFailedException e) {
// FIXME: 16/8/4 logger日志输出
e.printStackTrace();
}
currentLength += 4 + dataLength;
if (currentLength >= dataPackage.length) {
break;
}
}
return serializeData;
}
}

8
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/peresistent/PersistenceThreadLauncher.java
View File

@ -4,9 +4,11 @@ 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();
if (Config.Persistence.MAX_DEAL_DATA_THREAD_NUMBER > 0) {
new RegisterPersistenceThread().start();
for (int i = 0; i < Config.Persistence.MAX_DEAL_DATA_THREAD_NUMBER; i++) {
new PersistenceThread(i).start();
}
}
}
}

7
skywalking-server/src/main/resources/config.properties
View File

@ -1,9 +1,10 @@
#采集服务器的端口
server.port=34000
server.max_deal_data_thread_number=1
server.failed_package_watching_time_windowss=300
server.max_watching_failed_package_size=200
#
buffer.buffer_deal_thread_number=1
#每个线程最大缓存数量
buffer.per_thread_max_buffer_number=1024
#无数据处理时轮询等待时间(单位:毫秒)
@ -21,6 +22,8 @@ buffer.write_data_failure_retry_interval = 10000
persistence.switch_file_wait_time=5000
#追加EOF标志位的线程数量
persistence.max_append_eof_flags_thread_number=1
#持久化线程个数
persistence.max_deal_data_thread_number=0
#偏移量注册文件的目录
registerpersistence.register_file_parent_directory=/tmp/skywalking/data/offset

51
skywalking-server/src/test/java/com/ai/cloud/skywalking/reciever/model/BufferDataAssistTest.java Normal file
View File

@ -0,0 +1,51 @@
package com.ai.cloud.skywalking.reciever.model;
import com.ai.cloud.skywalking.protocol.AckSpan;
import com.ai.cloud.skywalking.protocol.RequestSpan;
import com.ai.cloud.skywalking.protocol.Span;
import com.ai.cloud.skywalking.protocol.TransportPackager;
import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable;
import com.ai.cloud.skywalking.protocol.common.ISerializable;
import com.ai.cloud.skywalking.protocol.common.SpanType;
import com.ai.cloud.skywalking.protocol.util.IntegerAssist;
import com.ai.cloud.skywalking.reciever.buffer.BufferDataAssist;
import com.ai.cloud.skywalking.reciever.peresistent.BufferFileReader;
import org.junit.Before;
import org.junit.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import static com.ai.cloud.skywalking.protocol.util.ByteDataUtil.unpackCheckSum;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
public class BufferDataAssistTest {
private List<ISerializable> serializableList;
@Test
public void pack() throws Exception {
byte[] byteData =
BufferDataAssist.appendLengthAndSplit(unpackCheckSum(TransportPackager.pack(serializableList)));
int length = IntegerAssist.bytesToInt(Arrays.copyOfRange(byteData, 0, 4), 0);
byte[] serializableByteData = Arrays.copyOfRange(byteData, 4, length + 4);
List<AbstractDataSerializable> serializables = BufferFileReader.deserializableObjects(serializableByteData);
assertEquals(2, serializables.size());
assertNotNull(serializables.get(0));
}
@Before
public void initData() {
serializableList = new ArrayList<ISerializable>();
Span span = new Span("test-traceID", "test", "10");
span.setStartDate(System.currentTimeMillis() - 1000 * 10);
span.setViewPointId("test-viewpoint");
span.setSpanType(SpanType.LOCAL);
serializableList.add(new RequestSpan(span));
serializableList.add(new AckSpan(span));
}
}