diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java index 4737b8008..faa4d636a 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java @@ -18,6 +18,7 @@ import io.netty.handler.codec.bytes.ByteArrayEncoder; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.Arrays; import com.ai.cloud.skywalking.selfexamination.HeathReading; import com.ai.cloud.skywalking.selfexamination.SDKHealthCollector; @@ -74,7 +75,14 @@ public class DataSender implements IDataSender { public boolean send(String data) { try { if (channel != null && channel.isActive()) { - channel.writeAndFlush(data.getBytes()); + // 对协议格式进行修改 + // | check sum(4 byte) | data + byte[] dataArray = new byte[data.getBytes().length + 4]; + System.arraycopy(dataArray,0, dataArray,4, dataArray.length); + byte[] checkSumArray = generateChecksum(data); + System.arraycopy(checkSumArray,0, dataArray,0, checkSumArray.length); + + channel.writeAndFlush(dataArray); SDKHealthCollector.getCurrentHeathReading("sender").updateData(HeathReading.INFO, "DataSender[" + socketAddress + "] send data successfully."); return true; }else{ @@ -89,6 +97,30 @@ public class DataSender implements IDataSender { return false; } + /** + * 生成校验和参数 + * @param data + * @return + */ + private byte[] generateChecksum(String data) { + char[] dataArray = data.toCharArray(); + int result = dataArray[0]; + for (int i = 0; i < dataArray.length; i++) { + result ^= dataArray[i]; + } + + return intToBytes(result); + } + + private byte[] intToBytes(int value) { + byte[] src = new byte[4]; + src[0] = (byte) ((value >> 24) & 0xFF); + src[1] = (byte) ((value >> 16) & 0xFF); + src[2] = (byte) ((value >> 8) & 0xFF); + src[3] = (byte) (value & 0xFF); + return src; + } + public InetSocketAddress getServerAddr() { return this.socketAddress; } diff --git a/skywalking-server/pom.xml b/skywalking-server/pom.xml index f7089f9a0..90b56db1b 100644 --- a/skywalking-server/pom.xml +++ b/skywalking-server/pom.xml @@ -4,7 +4,7 @@ com.ai.cloud skywalking-server - 1.0-Final + 1.0b jar skywalking-server @@ -14,6 +14,13 @@ UTF-8 + + + bintray-wu-sheng-sky-walking-repository + https://jcenter.bintray.com + + + junit @@ -44,7 +51,7 @@ com.ai.cloud skywalking-protocol - 1.0-Final + 1.0b redis.clients diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java index f4b262ea9..98e7c6683 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java @@ -8,6 +8,12 @@ public class Config { public static int PORT = 34000; // 最大数据处理线程数量 public static int MAX_DEAL_DATA_THREAD_NUMBER = 3; + + // 异常数据的时间间隔 + public static int EXCEPTION_DATA_SENDING_INTERVAL = 5 * 60; + + // 时间间隔内最大异常数据次数 + public static int MAX_SEND_EXCEPTION_DATA_COUNT = 200; } // 数据缓存配置类 @@ -86,19 +92,22 @@ public class Config { public static String STORAGE_TYPE = "hbase"; } - public static class Alarm { + public static class Redis{ - public static int ALARM_EXPIRE_SECONDS = 1000 * 60 * 90; - - public static int ALARM_EXCEPTION_STACK_LENGTH = 300; - - public static String REDIS_SERVER = "127.0.0.1:6379"; + public static String REDIS_SERVER = "10.1.241.18:16379"; public static int REDIS_MAX_IDLE = 10; public static int REDIS_MIN_IDLE = 1; 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; @@ -115,4 +124,5 @@ public class Config { // 默认健康检查上报时间 public static long REPORT_INTERVAL = 5 * 60 * 1000L; } + } \ No newline at end of file diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java index e660fdf39..8492a3510 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/handler/CollectionServerDataHandler.java @@ -2,8 +2,13 @@ package com.ai.cloud.skywalking.reciever.handler; import com.ai.cloud.skywalking.reciever.buffer.DataBufferThreadContainer; import com.ai.cloud.skywalking.reciever.conf.Config; +import com.ai.cloud.skywalking.reciever.storage.AlarmRedisConnector; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; +import redis.clients.jedis.Jedis; + +import java.net.InetSocketAddress; +import java.util.Arrays; public class CollectionServerDataHandler extends SimpleChannelInboundHandler { @@ -12,7 +17,56 @@ public class CollectionServerDataHandler extends SimpleChannelInboundHandler= 0 && msg.length < Config.DataPackage.MAX_DATA_PACKAGE) { - DataBufferThreadContainer.getDataBufferThread().saveTemporarily(msg); + // | check sum(4 byte) | data | + byte[] originCheckSum = new byte[4]; + System.arraycopy(msg, 0, originCheckSum, 0, 4); + // 对协议进行拆包 + String data = new String(msg, 4, msg.length - 4); + // 计算校验和 + byte[] checkSum = generateChecksum(data); + + if (Arrays.equals(originCheckSum, checkSum)) { + DataBufferThreadContainer.getDataBufferThread().saveTemporarily(data.getBytes()); + } else { + InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().localAddress(); + String key = socketAddress.getHostName() + ":" + socketAddress.getPort(); + Jedis jedis = AlarmRedisConnector.getJedis(); + // 如果不存在,则置为0,并且设置生失效时间 + if (jedis.setnx(key, 0 + "") == 1) { + jedis.expire(key, Config.Server.EXCEPTION_DATA_SENDING_INTERVAL); + } + + if (Config.Server.MAX_SEND_EXCEPTION_DATA_COUNT > jedis.incr(key)) { + ctx.channel().close(); + } + + } } } + + + /** + * 生成校验和参数 + * + * @param data + * @return + */ + private byte[] generateChecksum(String data) { + char[] dataArray = data.toCharArray(); + int result = dataArray[0]; + for (int i = 0; i < dataArray.length; i++) { + result ^= dataArray[i]; + } + + return intToBytes(result); + } + + private byte[] intToBytes(int value) { + byte[] src = new byte[4]; + src[0] = (byte) ((value >> 24) & 0xFF); + src[1] = (byte) ((value >> 16) & 0xFF); + src[2] = (byte) ((value >> 8) & 0xFF); + src[3] = (byte) (value & 0xFF); + return src; + } } diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/AlarmRedisConnector.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/AlarmRedisConnector.java index 3d6ff1e7d..c515a8fbc 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/AlarmRedisConnector.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/AlarmRedisConnector.java @@ -46,7 +46,7 @@ public class AlarmRedisConnector { public RedisInspector() { super("RedisInspectorThread"); - String redisServerConfig = Config.Alarm.REDIS_SERVER; + String redisServerConfig = Config.Redis.REDIS_SERVER; if (redisServerConfig == null || redisServerConfig.length() <= 0) { logger.error("Redis server is not setting. Switch off alarm module. "); Config.Alarm.ALARM_OFF_FLAG = true; @@ -75,7 +75,7 @@ public class AlarmRedisConnector { needConnectInit = false; } catch (Exception e) { logger.error("can't connect to redis[" - + Config.Alarm.REDIS_SERVER + "]", e); + + Config.Redis.REDIS_SERVER + "]", e); } finally { if (jedis != null) { jedis.close(); @@ -119,9 +119,9 @@ public class AlarmRedisConnector { private GenericObjectPoolConfig buildGenericObjectPoolConfig() { GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig(); genericObjectPoolConfig.setTestOnBorrow(true); - genericObjectPoolConfig.setMaxIdle(Config.Alarm.REDIS_MAX_IDLE); - genericObjectPoolConfig.setMinIdle(Config.Alarm.REDIS_MIN_IDLE); - genericObjectPoolConfig.setMaxTotal(Config.Alarm.REDIS_MAX_TOTAL); + genericObjectPoolConfig.setMaxIdle(Config.Redis.REDIS_MAX_IDLE); + genericObjectPoolConfig.setMinIdle(Config.Redis.REDIS_MIN_IDLE); + genericObjectPoolConfig.setMaxTotal(Config.Redis.REDIS_MAX_TOTAL); return genericObjectPoolConfig; } } diff --git a/skywalking-server/src/main/resources/config.properties b/skywalking-server/src/main/resources/config.properties index 9442627e0..693a95346 100644 --- a/skywalking-server/src/main/resources/config.properties +++ b/skywalking-server/src/main/resources/config.properties @@ -1,6 +1,8 @@ #采集服务器的端口 server.port=34000 server.max_deal_data_thread_number=5 +server.exception_data_sending_interval=300 +server.max_send_exception_data_count=200; #每个线程最大缓存数量 buffer.per_thread_max_buffer_number=1024 @@ -51,19 +53,20 @@ hbaseconfig.client_port=29181 #告警失效时间 alarm.alarm_expire_seconds=5400 -#Redis配置 -alarm.redis_server=10.1.241.18:16379 -#Redis最大空闲数量 -alarm.edis_max_idle=10 -#Redis最小空闲数量 -alarm.edis_min_idle=1 -#Redis最大个数 -alarm.edis_max_total=20 #是否关闭告警 alarm.larm_off_flag=false #告警redis检测器检测周期 alarm.alarm_redis_inspector_interval=5000 +#Redis配置 +redis.redis_server=10.1.241.18:16379 +#Redis最大空闲数量 +redis.edis_max_idle=10 +#Redis最小空闲数量 +redis.edis_min_idle=1 +#Redis最大个数 +redis.edis_max_total=20 + #告警检查器:异常告警检查 alarm.checker.turn_on_exception_checker=true #告警检查器:执行时间超时告警检查