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

1.修改BusinessKeyAppender的setBusinessKey2Trace方法 

2.增加客户端的日志收集（还缺少上报程序），包为：com.ai.cloud.skywalking.selfexamination
3.移除无用的constant包，原有类更换包名。
This commit is contained in:
wusheng 2015-11-23 16:37:26 +08:00
parent f946dfbcab
commit 406d4e112c
16 changed files with 161 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
skywalking-api/src/main/java/com/ai/cloud/skywalking/api/BusinessKeyAppender.java
View File

@ -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;

17
skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java
View File

@ -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();
}

4
skywalking-api/src/main/java/com/ai/cloud/skywalking/constants/Constants.java → skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Constants.java
View File

@ -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 = "^~";
}

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

@ -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 {

34
skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HealthCollector.java Normal file
View File

@ -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<String, HeathReading> heathReadings = new HashMap<String, HeathReading>();
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: 使用专有的端口，将数据上报给服务端，定时上报，默认应为分钟级别，降低服务端压力
}
}

74
skywalking-api/src/main/java/com/ai/cloud/skywalking/selfexamination/HeathReading.java Normal file
View File

@ -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<String, HeathDetailData> datas = new HashMap<String, HeathDetailData>();
/**
* 健康读数，只应该在工作线程中创建
*
*/
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<String, HeathDetailData> 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<String, HeathReading.HeathDetailData>();
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;
}
}
}

48
skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java
View File

@ -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<SocketAddress> socketAddresses = new HashSet<SocketAddress>();
private static Set<SocketAddress> unUsedSocketAddresses = new HashSet<SocketAddress>();
@ -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");
}
}
}

4
skywalking-api/src/main/java/com/ai/cloud/skywalking/util/BuriedPointMachineUtil.java
View File

@ -52,6 +52,10 @@ public final class BuriedPointMachineUtil {
}
return hostName;
}
public static String getHostDesc(){
return getHostName() + "/" + getHostIp();
}
private BuriedPointMachineUtil() {
// Non

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

@ -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() {

2
skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ExceptionHandleUtil.java
View File

@ -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;

2
skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainE.java
View File

@ -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));
}

2
skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainG.java
View File

@ -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));

2
skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainH.java
View File

@ -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));

2
skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainI.java
View File

@ -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));

2
skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainJ.java
View File

@ -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));

2
skywalking-sdk-plugin/spring-plugin/src/test/java/com/ai/cloud/skywalking/plugin/spring/common/CallChainK.java
View File

@ -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));
}
}