diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/api/BusinessKeyAppender.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/api/BusinessKeyAppender.java index f135c9e45..2b9d7f414 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/api/BusinessKeyAppender.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/api/BusinessKeyAppender.java @@ -16,7 +16,7 @@ public final class BusinessKeyAppender { * * @param businessKey */ - public static void trace(String businessKey) { + public static void setBusinessKey2Trace(String businessKey) { if (!AuthDesc.isAuth()) return; diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java index 6a8c55a1e..c849e12d5 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java @@ -3,6 +3,8 @@ package com.ai.cloud.skywalking.buffer; import com.ai.cloud.skywalking.conf.Config; import com.ai.cloud.skywalking.context.Span; +import com.ai.cloud.skywalking.selfexamination.HealthCollector; +import com.ai.cloud.skywalking.selfexamination.HeathReading; import com.ai.cloud.skywalking.sender.DataSenderFactory; import java.util.concurrent.atomic.AtomicInteger; @@ -23,10 +25,9 @@ public class BufferGroup { int step = (int) Math.ceil(BUFFER_MAX_SIZE * 1.0 / MAX_CONSUMER); int start = 0, end = 0; - int i = 1; while (true) { if (end + step >= BUFFER_MAX_SIZE) { - new ConsumerWorker(groupName + "-consumer-" + (i++), start, BUFFER_MAX_SIZE).start(); + new ConsumerWorker(start, BUFFER_MAX_SIZE).start(); break; } end += step; @@ -38,8 +39,7 @@ public class BufferGroup { public void save(Span span) { int i = Math.abs(index.getAndIncrement() % BUFFER_MAX_SIZE); if (dataBuffer[i] != null) { - // TODO 需要上报 - System.out.println(span.getLevelId() + "在Group[" + groupName + "]的第" + i + "位冲突"); + HealthCollector.getCurrentHeathReading(null).updateData(HeathReading.WARNING, span.getLevelId() + "在Group[" + groupName + "]的第" + i + "位冲突"); } dataBuffer[i] = span; } @@ -49,17 +49,11 @@ public class BufferGroup { private int end = BUFFER_MAX_SIZE; private ConsumerWorker(int start, int end) { + super("ConsumerWorker"); this.start = start; this.end = end; } - private ConsumerWorker(String threadName, int start, int end) { - super(threadName); - this.start = start; - this.end = end; - } - - @Override public void run() { StringBuilder data = new StringBuilder(); @@ -78,6 +72,7 @@ public class BufferGroup { logger.log(Level.ALL, "Sleep Failure"); } } + HealthCollector.getCurrentHeathReading(null).updateData(HeathReading.INFO, "send buried-point data."); data = new StringBuilder(); } diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/constants/Constants.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Constants.java similarity index 72% rename from skywalking-api/src/main/java/com/ai/cloud/skywalking/constants/Constants.java rename to skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Constants.java index 5418df294..5b609dd8d 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/constants/Constants.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Constants.java @@ -1,4 +1,4 @@ -package com.ai.cloud.skywalking.constants; +package com.ai.cloud.skywalking.conf; public class Constants { @@ -9,4 +9,6 @@ public class Constants { public static final String NEW_LINE_CHARACTER_PATTERN = "\\n"; public static final String EXCEPTION_SPILT_PATTERN = "^"; + + public static final String HEALTH_DATA_SPILT_PATTERN = "^~"; } diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/Span.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/Span.java index 3f81667a6..ddf9333cf 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/Span.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/context/Span.java @@ -1,6 +1,6 @@ package com.ai.cloud.skywalking.context; -import com.ai.cloud.skywalking.constants.Constants; +import com.ai.cloud.skywalking.conf.Constants; import com.ai.cloud.skywalking.util.StringUtil; public class Span { diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HealthCollector.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HealthCollector.java new file mode 100644 index 000000000..9655c0192 --- /dev/null +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HealthCollector.java @@ -0,0 +1,34 @@ +package com.ai.cloud.skywalking.selfexamination; + +import java.util.HashMap; +import java.util.Map; + +import static com.ai.cloud.skywalking.conf.Config.SkyWalking.USER_ID; +import com.ai.cloud.skywalking.util.BuriedPointMachineUtil; + +public class HealthCollector extends Thread{ + private static Map heathReadings = new HashMap(); + + public static HeathReading getCurrentHeathReading(String extraId){ + String id = getId(extraId); + if(!heathReadings.containsKey(id)){ + synchronized (heathReadings) { + if(!heathReadings.containsKey(id)){ + heathReadings.put(id, new HeathReading(id)); + } + } + } + return heathReadings.get(id); + } + + private static String getId(String extraId){ + return "SDK,U:" + USER_ID + ",M:" + BuriedPointMachineUtil.getHostDesc() +",P:" + BuriedPointMachineUtil.getProcessNo() + ",T:" + + Thread.currentThread().getName() + "(" + + Thread.currentThread().getId() + ")" + (extraId == null? "" : ",extra:" + extraId); + } + + @Override + public void run(){ + //TODO: 使用专有的端口，将数据上报给服务端，定时上报，默认应为分钟级别，降低服务端压力 + } +} diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HeathReading.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HeathReading.java new file mode 100644 index 000000000..1d7a89476 --- /dev/null +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HeathReading.java @@ -0,0 +1,74 @@ +package com.ai.cloud.skywalking.selfexamination; + +import java.util.HashMap; +import java.util.Map; + +import com.ai.cloud.skywalking.conf.Constants; + +public class HeathReading { + public static final String ERROR = "ERROR"; + public static final String WARNING = "WARNING"; + public static final String INFO = "INFO"; + + private String id; + + private Map datas = new HashMap(); + + /** + * 健康读数，只应该在工作线程中创建 + * + */ + public HeathReading(String id) { + this.id = id; + } + + public void updateData(String key, String newData){ + if(datas.containsKey(key)){ + datas.get(key).updateData(newData); + }else{ + datas.put(key, new HeathDetailData(newData)); + } + } + + @Override + public String toString(){ + StringBuilder sb = new StringBuilder(this.id); + sb.append(Constants.HEALTH_DATA_SPILT_PATTERN); + for(Map.Entry data : datas.entrySet()){ + sb.append(data.getKey()).append(Constants.HEALTH_DATA_SPILT_PATTERN).append(data.getValue().toString()).append(Constants.HEALTH_DATA_SPILT_PATTERN); + } + + //reset data + datas = new HashMap(); + return sb.toString(); + } + + class HeathDetailData{ + private String data; + + private long statusTime; + + HeathDetailData(String initialData){ + data = initialData; + statusTime = System.currentTimeMillis(); + } + + void updateData(String newData){ + data = newData; + statusTime = System.currentTimeMillis(); + } + + String getData() { + return data; + } + + long getStatusTime() { + return statusTime; + } + + @Override + public String toString(){ + return "d:" + data + Constants.HEALTH_DATA_SPILT_PATTERN + "t:" + statusTime; + } + } +} diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java index d6ac3172f..be1559627 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java @@ -1,23 +1,29 @@ package com.ai.cloud.skywalking.sender; -import com.ai.cloud.skywalking.conf.Config; -import com.ai.cloud.skywalking.util.StringUtil; - -import java.io.IOException; -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.util.*; -import java.util.concurrent.ThreadLocalRandom; -import java.util.logging.Level; -import java.util.logging.Logger; - import static com.ai.cloud.skywalking.conf.Config.Sender.CONNECT_PERCENT; import static com.ai.cloud.skywalking.conf.Config.Sender.RETRY_GET_SENDER_WAIT_INTERVAL; import static com.ai.cloud.skywalking.conf.Config.SenderChecker.CHECK_POLLING_TIME; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; +import java.util.logging.Level; +import java.util.logging.Logger; + +import com.ai.cloud.skywalking.conf.Config; +import com.ai.cloud.skywalking.selfexamination.HealthCollector; +import com.ai.cloud.skywalking.selfexamination.HeathReading; +import com.ai.cloud.skywalking.util.StringUtil; + public class DataSenderFactory { - //private static Logger logger = Logger.getLogger(DataSenderFactory.getSender().toString()); + private static Logger logger = Logger.getLogger(DataSenderFactory.getSender().toString()); private static Set socketAddresses = new HashSet(); private static Set unUsedSocketAddresses = new HashSet(); @@ -37,7 +43,7 @@ public class DataSenderFactory { socketAddresses.add(new InetSocketAddress(server[0], Integer.valueOf(server[1]))); } } catch (Exception e) { - // logger.log(Level.ALL, "Collection service configuration error."); + logger.log(Level.ALL, "Collection service configuration error.", e); System.exit(-1); } @@ -49,7 +55,7 @@ public class DataSenderFactory { try { Thread.sleep(RETRY_GET_SENDER_WAIT_INTERVAL); } catch (InterruptedException e) { - // logger.log(Level.ALL, "Sleep failure"); + logger.log(Level.ALL, "Sleep failure", e); } } return availableSenders.get(ThreadLocalRandom.current().nextInt(0, availableSenders.size())); @@ -60,8 +66,10 @@ public class DataSenderFactory { private int availableSize; public DataSenderChecker() { + super("DataSenderChecker"); + if (CONNECT_PERCENT <= 0 || CONNECT_PERCENT > 100) { - // logger.log(Level.ALL, "CONNECT_PERCENT must between 1 and 100"); + logger.log(Level.ALL, "CONNECT_PERCENT must between 1 and 100"); System.exit(-1); } availableSize = (int) Math.ceil(socketAddresses.size() * ((1.0 * CONNECT_PERCENT / 100) % 100)); @@ -92,25 +100,29 @@ public class DataSenderFactory { tmpScoketAddress = unUsedSocketAddressIterator.next(); if (availableSenders.size() >= availableSize) { + HealthCollector.getCurrentHeathReading(null).updateData(HeathReading.INFO, "the num of available senders is enough."); break; } synchronized (lock) { try { - + HealthCollector.getCurrentHeathReading(null).updateData(HeathReading.INFO, "increasing available senders."); availableSenders.add(new DataSender(tmpScoketAddress)); unUsedSocketAddresses.remove(tmpScoketAddress); } catch (IOException e) { } } - + } + + if (availableSenders.size() >= availableSize) { + HealthCollector.getCurrentHeathReading(null).updateData(HeathReading.WARNING, "the num of available senders is not enough (" + availableSenders.size() + ")."); } try { Thread.sleep(CHECK_POLLING_TIME); } catch (InterruptedException e) { - //logger.log(Level.ALL, "Sleep Failure"); + logger.log(Level.ALL, "Sleep Failure"); } } } diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/BuriedPointMachineUtil.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/BuriedPointMachineUtil.java index 71537327d..bfd9fc661 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/BuriedPointMachineUtil.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/BuriedPointMachineUtil.java @@ -52,6 +52,10 @@ public final class BuriedPointMachineUtil { } return hostName; } + + public static String getHostDesc(){ + return getHostName() + "/" + getHostIp(); + } private BuriedPointMachineUtil() { // Non diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ContextGenerator.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ContextGenerator.java index e36c61ced..83b04950e 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ContextGenerator.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ContextGenerator.java @@ -48,7 +48,7 @@ public final class ContextGenerator { // 设置基本信息 spanData.setStartDate(System.currentTimeMillis()); spanData.setProcessNo(BuriedPointMachineUtil.getProcessNo()); - spanData.setAddress(BuriedPointMachineUtil.getHostName() + "/" + BuriedPointMachineUtil.getHostIp()); + spanData.setAddress(BuriedPointMachineUtil.getHostDesc()); } private static Span getSpanFromThreadLocal() { diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ExceptionHandleUtil.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ExceptionHandleUtil.java index 21026ca0f..5df383102 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ExceptionHandleUtil.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ExceptionHandleUtil.java @@ -1,6 +1,6 @@ package com.ai.cloud.skywalking.util; -import com.ai.cloud.skywalking.constants.Constants; +import com.ai.cloud.skywalking.conf.Constants; import com.ai.cloud.skywalking.context.Context; import com.ai.cloud.skywalking.context.Span; diff --git a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainE.java b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainE.java index 0a7125e75..06aad71fb 100644 --- a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainE.java +++ b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainE.java @@ -12,7 +12,7 @@ public class CallChainE { @Tracing public void doBusiness() throws InterruptedException { Thread.sleep(ThreadLocalRandom.current().nextInt(10)); - BusinessKeyAppender.trace("key-value"); + BusinessKeyAppender.setBusinessKey2Trace("key-value"); callChainG.doBusiness(); Thread.sleep(ThreadLocalRandom.current().nextInt(10)); } diff --git a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainG.java b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainG.java index 24186d837..eca0406a1 100644 --- a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainG.java +++ b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainG.java @@ -12,7 +12,7 @@ public class CallChainG { @Tracing public void doBusiness() throws InterruptedException { Thread.sleep(ThreadLocalRandom.current().nextInt(10)); - BusinessKeyAppender.trace("key-value"); + BusinessKeyAppender.setBusinessKey2Trace("key-value"); callChainH.doBusiness(); Thread.sleep(ThreadLocalRandom.current().nextInt(10)); diff --git a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainH.java b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainH.java index 04de8bbcf..e905defad 100644 --- a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainH.java +++ b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainH.java @@ -12,7 +12,7 @@ public class CallChainH { @Tracing public void doBusiness() throws InterruptedException { Thread.sleep(ThreadLocalRandom.current().nextInt(10)); - BusinessKeyAppender.trace("key-value"); + BusinessKeyAppender.setBusinessKey2Trace("key-value"); callChainI.doBusiness(); Thread.sleep(ThreadLocalRandom.current().nextInt(10)); diff --git a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainI.java b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainI.java index 37074d455..4d825cc38 100644 --- a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainI.java +++ b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainI.java @@ -11,7 +11,7 @@ public class CallChainI { @Tracing public void doBusiness() throws InterruptedException { Thread.sleep(ThreadLocalRandom.current().nextInt(10)); - BusinessKeyAppender.trace("key-value"); + BusinessKeyAppender.setBusinessKey2Trace("key-value"); callChainJ.doBusiness(); Thread.sleep(ThreadLocalRandom.current().nextInt(10)); diff --git a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainJ.java b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainJ.java index 8f1828f80..a33e3fc73 100644 --- a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainJ.java +++ b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainJ.java @@ -10,7 +10,7 @@ public class CallChainJ { @Tracing public void doBusiness() throws InterruptedException { Thread.sleep(ThreadLocalRandom.current().nextInt(10)); - BusinessKeyAppender.trace("key-value"); + BusinessKeyAppender.setBusinessKey2Trace("key-value"); callChainK.doBusiness(); Thread.sleep(ThreadLocalRandom.current().nextInt(10)); diff --git a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainK.java b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainK.java index 8516ac318..d440c1b6e 100644 --- a/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainK.java +++ b/skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainK.java @@ -9,7 +9,7 @@ public class CallChainK { @Tracing public void doBusiness() throws InterruptedException { Thread.sleep(ThreadLocalRandom.current().nextInt(10)); - BusinessKeyAppender.trace("key-value"); + BusinessKeyAppender.setBusinessKey2Trace("key-value"); Thread.sleep(ThreadLocalRandom.current().nextInt(10)); } }