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

1. 增加配置 

2. 增加log
This commit is contained in:
zhangxin10 2015-11-14 14:52:52 +08:00
parent d9fec09599
commit 64f672ec37
14 changed files with 135 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -8,6 +8,8 @@ sender.read_step_length=256
consumer.max_consumer=1
#消费者最大等待时间
consumer.max_wait_time=5
#发送失败等待时间
consumer.consumer_fail_retry_wait_interval=50
#每个Buffer的最大个数
buffer.buffer_max_size=18000
#Buffer池的最大长度
@ -16,8 +18,14 @@ buffer.pool_max_length=1
sender.max_buffer_data_size=10
#最大发送者的连接数阀比例
sender.connect_percent=100
#当没有Sender时,尝试获取sender的等待周期
sender.retry_get_sender_wait_interval=2000
#是否开启发送消息
sender.is_off=false
#发送服务端配置
#sender.servers_addr=10.1.235.197:34000;10.1.235.197:35000
sender.servers_addr=127.0.0.1:34000;127.0.0.1:35000
sender.servers_addr=127.0.0.1:34000;127.0.0.1:35000
buriedpoint.max_exception_stack_length=4000
#发送检查线程检查周期
senderchecker.check_polling_time=200

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

@ -5,16 +5,16 @@ import com.ai.cloud.skywalking.conf.Config;
import com.ai.cloud.skywalking.context.Span;
import com.ai.cloud.skywalking.sender.DataSenderFactory;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;
import static com.ai.cloud.skywalking.conf.Config.Buffer.BUFFER_MAX_SIZE;
import static com.ai.cloud.skywalking.conf.Config.Consumer.MAX_CONSUMER;
import static com.ai.cloud.skywalking.conf.Config.Consumer.MAX_WAIT_TIME;
import static com.ai.cloud.skywalking.conf.Config.Consumer.*;
import static com.ai.cloud.skywalking.conf.Config.Sender.MAX_BUFFER_DATA_SIZE;
public class BufferGroup {
public static CountDownLatch count;
private static Logger logger = Logger.getLogger(BufferGroup.class.getName());
private String groupName;
private Span[] dataBuffer = new Span[BUFFER_MAX_SIZE];
AtomicInteger index = new AtomicInteger(0);
@ -24,9 +24,10 @@ 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(start, BUFFER_MAX_SIZE).start();
new ConsumerWorker(groupName + "-consumer-" + (i++), start, BUFFER_MAX_SIZE).start();
break;
}
end += step;
@ -53,9 +54,13 @@ public class BufferGroup {
this.end = end;
}
ConsumerWorker() {
private ConsumerWorker(String threadName, int start, int end) {
super(threadName);
this.start = start;
this.end = end;
}
@Override
public void run() {
int index = 0;
@ -72,9 +77,9 @@ public class BufferGroup {
if (index++ == MAX_BUFFER_DATA_SIZE || data.length() >= Config.Sender.MAX_SEND_LENGTH) {
while (!DataSenderFactory.getSender().send(data.toString())) {
try {
Thread.sleep(50L);
Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL);
} catch (InterruptedException e) {
e.printStackTrace();
logger.log(Level.ALL, "Sleep Failure");
}
}
index = 0;
@ -86,7 +91,7 @@ public class BufferGroup {
try {
Thread.sleep(MAX_WAIT_TIME);
} catch (InterruptedException e) {
e.printStackTrace();
logger.log(Level.ALL, "Sleep Failure");
}
}
}

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

@ -11,7 +11,7 @@ class BufferPool {
static {
for (int i = 0; i < POOL_MAX_LENGTH; i++) {
bufferGroups[i] = new BufferGroup("BufferLine-" + i);
bufferGroups[i] = new BufferGroup("buffer_group-" + i);
}
}

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

@ -7,14 +7,18 @@ import com.ai.cloud.skywalking.context.Span;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.logging.Level;
import java.util.logging.Logger;
public class ContextBuffer {
private static Logger logger = Logger.getLogger(ContextBuffer.class.getName());
private static boolean isAuth = true;
static {
InputStream inputStream = ContextBuffer.class.getResourceAsStream("/sky-walking.auth");
if (inputStream == null) {
isAuth = false;
logger.log(Level.ALL, "No provider sky-walking certification documents, buried point won't work");
}
if (isAuth) {
try {
@ -23,8 +27,10 @@ public class ContextBuffer {
ConfigInitializer.initialize(properties, Config.class);
} catch (IllegalAccessException e) {
isAuth = false;
logger.log(Level.ALL, "Parsing certification file failed, buried won't work");
} catch (IOException e) {
isAuth = false;
logger.log(Level.ALL, "Failed to read the certification file, buried won't work");
}
ContextBuffer.init();

16
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/LocalBuriedPointSender.java
View File

@ -10,13 +10,18 @@ import com.ai.cloud.skywalking.model.Identification;
import com.ai.cloud.skywalking.util.ContextGenerator;
import com.ai.cloud.skywalking.util.ExceptionHandleUtil;
import java.util.logging.Level;
import java.util.logging.Logger;
public class LocalBuriedPointSender implements IBuriedPointSender {
private static Logger logger = Logger.getLogger(LocalBuriedPointSender.class.getName());
public ContextData beforeSend(Identification id) {
Span spanData = ContextGenerator.generateSpanFromThreadLocal(id);
// 3.将新创建的Context存放到ThreadLocal栈中。
// 将新创建的Context存放到ThreadLocal栈中。
Context.append(spanData);
// 4 并将当前的Context返回回去
// 并将当前的Context返回回去
return new ContextData(spanData);
}
@ -26,12 +31,15 @@ public class LocalBuriedPointSender implements IBuriedPointSender {
if (spanData == null) {
return;
}
// 填上必要信息
// 加上花费时间
spanData.setCost(System.currentTimeMillis() - spanData.getStartDate());
if (Config.BuriedPoint.PRINTF) {
System.out.println("viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
logger.log(Level.INFO, "viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
getParentLevel() + "\tLevelId:" + spanData.getLevelId());
}
// 存放到本地发送进程中
if (!Config.Sender.IS_OFF) {
ContextBuffer.save(spanData);

11
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/RPCBuriedPointReceiver.java
View File

@ -10,8 +10,13 @@ import com.ai.cloud.skywalking.model.Identification;
import com.ai.cloud.skywalking.util.ContextGenerator;
import com.ai.cloud.skywalking.util.ExceptionHandleUtil;
import java.util.logging.Level;
import java.util.logging.Logger;
public class RPCBuriedPointReceiver implements IBuriedPointReceiver {
private static Logger logger = Logger.getLogger(LocalBuriedPointSender.class.getName());
public void afterReceived() {
// 获取上下文的栈顶中的元素
Span spanData = Context.removeLastSpan();
@ -23,12 +28,14 @@ public class RPCBuriedPointReceiver implements IBuriedPointReceiver {
public void beforeReceived(ContextData context, Identification id) {
Span spanData = ContextGenerator.generateSpanFromContextData(context, id);
//设置是否为接收端
spanData.setReceiver(true);
// 存放到上下文
if (Config.BuriedPoint.PRINTF) {
System.out.println("viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
logger.log(Level.INFO, "viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
getParentLevel() + "\tLevelId:" + spanData.getLevelId());
}
Context.append(spanData);
}

18
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/RPCBuriedPointSender.java
View File

@ -10,29 +10,35 @@ import com.ai.cloud.skywalking.model.Identification;
import com.ai.cloud.skywalking.util.ContextGenerator;
import com.ai.cloud.skywalking.util.ExceptionHandleUtil;
import java.util.logging.Level;
import java.util.logging.Logger;
public class RPCBuriedPointSender implements IBuriedPointSender {
private static Logger logger = Logger.getLogger(RPCBuriedPointReceiver.class.getName());
public ContextData beforeSend(Identification id) {
Span spanData = ContextGenerator.generateSpanFromThreadLocal(id);
// 3.将新创建的Context存放到ThreadLocal栈中。
// 将新创建的Context存放到ThreadLocal栈中。
Context.append(spanData);
// 4 并将当前的Context返回回去
// 并将当前的Context返回回去
return new ContextData(spanData);
}
public void afterSend() {
// 获取上下文的栈顶中的元素
Span spanData = Context.removeLastSpan();
Span spanData = Context.removeLastSpan();
if (spanData == null) {
return;
}
// 填上必要信息
spanData.setCost(System.currentTimeMillis() - spanData.getStartDate());
if (Config.BuriedPoint.PRINTF) {
System.out.println("viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
logger.log(Level.ALL, "viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
getParentLevel() + "\tLevelId:" + spanData.getLevelId());
}
// 存放到本地发送进程中
if (!Config.Sender.IS_OFF) {
ContextBuffer.save(spanData);
}

7
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/ThreadBuriedPointSender.java
View File

@ -11,8 +11,13 @@ import com.ai.cloud.skywalking.util.BuriedPointMachineUtil;
import com.ai.cloud.skywalking.util.ExceptionHandleUtil;
import com.ai.cloud.skywalking.util.TraceIdGenerator;
import java.util.logging.Level;
import java.util.logging.Logger;
public class ThreadBuriedPointSender implements IBuriedPointSender {
private static Logger logger = Logger.getLogger(ThreadBuriedPointSender.class.getName());
private Span span;
public ThreadBuriedPointSender(int threadSeqId) {
@ -49,7 +54,7 @@ public class ThreadBuriedPointSender implements IBuriedPointSender {
// 填上必要信息
span.setCost(System.currentTimeMillis() - span.getStartDate());
if (Config.BuriedPoint.PRINTF) {
System.out.println("viewpointId:" + span.getViewPointId() + "\tParentLevelId:" + span.
logger.log(Level.INFO, "viewpointId:" + span.getViewPointId() + "\tParentLevelId:" + span.
getParentLevel() + "\tLevelId:" + span.getLevelId());
}
// 存放到本地发送进程中

11
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/ThreadFactoryBuriedPointSender.java
View File

@ -10,13 +10,18 @@ import com.ai.cloud.skywalking.model.Identification;
import com.ai.cloud.skywalking.util.ContextGenerator;
import com.ai.cloud.skywalking.util.ExceptionHandleUtil;
import java.util.logging.Level;
import java.util.logging.Logger;
public class ThreadFactoryBuriedPointSender implements IBuriedPointSender {
private static Logger logger = Logger.getLogger(ThreadBuriedPointSender.class.getName());
public ContextData beforeSend(Identification id) {
Span spanData = ContextGenerator.generateSpanFromThreadLocal(id);
// 3.将新创建的Context存放到ThreadLocal栈中。
// 将新创建的Context存放到ThreadLocal栈中。
Context.append(spanData);
// 4 并将当前的Context返回回去
// 并将当前的Context返回回去
return new ContextData(spanData);
}
@ -29,7 +34,7 @@ public class ThreadFactoryBuriedPointSender implements IBuriedPointSender {
// 填上必要信息
spanData.setCost(System.currentTimeMillis() - spanData.getStartDate());
if (Config.BuriedPoint.PRINTF) {
System.out.println("viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
logger.log(Level.INFO, "viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
getParentLevel() + "\tLevelId:" + spanData.getLevelId());
}
// 存放到本地发送进程中

18
skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java
View File

@ -3,7 +3,10 @@ package com.ai.cloud.skywalking.conf;
public class Config {
public static class BuriedPoint {
//是否打印埋点信息
public static boolean PRINTF = false;
public static int MAX_EXCEPTION_STACK_LENGTH = 4000;
}
public static class Consumer {
@ -11,6 +14,9 @@ public class Config {
public static int MAX_CONSUMER = 2;
//消费者最大等待时间
public static long MAX_WAIT_TIME = 5L;
//
public static long CONSUMER_FAIL_RETRY_WAIT_INTERVAL = 50L;
}
public static class Buffer {
@ -36,5 +42,17 @@ public class Config {
// 发送的最大长度
public static int MAX_SEND_LENGTH = 1800;
public static long RETRY_GET_SENDER_WAIT_INTERVAL = 2000L;
}
public static class SenderChecker {
//检查周期时间
public static long CHECK_POLLING_TIME = 200L;
}
}

3
skywalking-api/src/main/java/com/ai/cloud/skywalking/context/Context.java
View File

@ -24,8 +24,9 @@ public class Context {
}
public static Span removeLastSpan() {
if (nodes.get() == null)
if (nodes.get() == null) {
return null;
}
return nodes.get().pop();
}

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

@ -8,15 +8,21 @@ 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;
public class DataSenderFactory {
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>();
private static List<DataSender> availableSenders = new ArrayList<DataSender>();
private static DataSenderMaker dataSenderMaker;
private static DataSenderChecker dataSenderChecker;
private static Object lock = new Object();
static {
@ -32,40 +38,40 @@ public class DataSenderFactory {
socketAddresses.add(new InetSocketAddress(server[0], Integer.valueOf(server[1])));
}
} catch (Exception e) {
System.err.print("Collection service configuration error.");
logger.log(Level.ALL, "Collection service configuration error.");
System.exit(-1);
}
dataSenderMaker = new DataSenderMaker();
dataSenderMaker.start();
dataSenderChecker = new DataSenderChecker();
dataSenderChecker.start();
}
public static DataSender getSender() {
while (availableSenders.size() == 0) {
try {
Thread.sleep(2000L);
Thread.sleep(RETRY_GET_SENDER_WAIT_INTERVAL);
} catch (InterruptedException e) {
e.printStackTrace();
logger.log(Level.ALL, "Sleep failure");
}
}
return availableSenders.get(ThreadLocalRandom.current().nextInt(0, availableSenders.size()));
}
static class DataSenderMaker extends Thread {
static class DataSenderChecker extends Thread {
private int avaiableSize;
private int availableSize;
public DataSenderMaker() {
public DataSenderChecker() {
if (CONNECT_PERCENT <= 0 || CONNECT_PERCENT > 100) {
System.err.println("CONNECT_PERCENT must between 1 and 100");
logger.log(Level.ALL, "CONNECT_PERCENT must between 1 and 100");
System.exit(-1);
}
avaiableSize = (int) Math.ceil(socketAddresses.size() * ((1.0 * CONNECT_PERCENT / 100) % 100));
availableSize = (int) Math.ceil(socketAddresses.size() * ((1.0 * CONNECT_PERCENT / 100) % 100));
// 初始化DataSender
List<SocketAddress> usedSocketAddress = new ArrayList<SocketAddress>();
for (SocketAddress socketAddress : socketAddresses) {
if (availableSenders.size() >= avaiableSize) {
if (availableSenders.size() >= availableSize) {
break;
}
try {
@ -74,28 +80,35 @@ public class DataSenderFactory {
} catch (IOException e) {
unUsedSocketAddresses.add(socketAddress);
}
}
unUsedSocketAddresses = new HashSet<SocketAddress>(socketAddresses);
unUsedSocketAddresses.removeAll(usedSocketAddress);
}
public void run() {
Iterator<SocketAddress> unUsedSocketAddressIterator;
SocketAddress tmpScoketAddress;
while (true) {
for (SocketAddress socketAddress : unUsedSocketAddresses) {
if (availableSenders.size() >= avaiableSize) {
unUsedSocketAddressIterator = unUsedSocketAddresses.iterator();
while (unUsedSocketAddressIterator.hasNext()) {
tmpScoketAddress = unUsedSocketAddressIterator.next();
if (availableSenders.size() >= availableSize) {
break;
}
try {
availableSenders.add(new DataSender(socketAddress));
availableSenders.add(new DataSender(tmpScoketAddress));
unUsedSocketAddresses.remove(tmpScoketAddress);
} catch (IOException e) {
// 当前发送的地址还是不可用
}
}
try {
Thread.sleep(200L);
Thread.sleep(CHECK_POLLING_TIME);
} catch (InterruptedException e) {
e.printStackTrace();
logger.log(Level.ALL, "Sleep Failure");
}
}
}
@ -106,7 +119,6 @@ public class DataSenderFactory {
availableSenders.remove(sender);
unUsedSocketAddresses.add(sender.getServerIp());
}
}
}

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

@ -9,7 +9,7 @@ public final class ContextGenerator {
/**
* 利用本地ThreadLocal的信息创建Context，主要用于非跨JVM的操作
*
* @param sendData 视点，业务数据等信息
* @param id 视点，业务数据等信息
* @return
*/
public static Span generateSpanFromThreadLocal(Identification id) {

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

@ -6,12 +6,14 @@ import com.ai.cloud.skywalking.context.Span;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import static com.ai.cloud.skywalking.conf.Config.BuriedPoint.MAX_EXCEPTION_STACK_LENGTH;
public final class ExceptionHandleUtil {
private static String extractExceptionStackMessage(final Throwable e) {
ByteArrayOutputStream buf = new ByteArrayOutputStream();
StringBuilder expMessage = new StringBuilder();
Throwable causeException = e;
while (causeException.getCause() != null || expMessage.length() < 40000) {
while (causeException.getCause() != null || expMessage.length() < MAX_EXCEPTION_STACK_LENGTH) {
causeException.printStackTrace(new java.io.PrintWriter(buf, true));
expMessage.append(buf.toString());
causeException = causeException.getCause();
@ -22,10 +24,10 @@ public final class ExceptionHandleUtil {
expMessage.append("\n本地发送埋点关闭异常读入流异常，异常信息:");
expMessage.append(e1.getCause().getMessage());
}
if (expMessage.length() <= 4000) {
if (expMessage.length() <= MAX_EXCEPTION_STACK_LENGTH) {
return expMessage.toString().replace('\n', '&');
} else {
return expMessage.toString().replace('\n', '&').substring(0, 40000);
return expMessage.toString().replace('\n', '&').substring(0, MAX_EXCEPTION_STACK_LENGTH);
}
}