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

1.修改服务端从缓冲文件读入数据的方式（从单个字符读入改成整行读入） 

2.插入HBase方式改成批量插入
This commit is contained in:
zhangxin10 2015-11-18 14:49:31 +08:00
parent 77f71d5cbb
commit 2c0d02907f
14 changed files with 233 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6,7 +6,6 @@ import com.ai.cloud.skywalking.context.Context;
import com.ai.cloud.skywalking.context.Span;
public final class BusinessKeyAppender {
private static final char spiltChar = '^';
private BusinessKeyAppender() {
// Non
@ -14,7 +13,7 @@ public final class BusinessKeyAppender {
/**
* 在当前埋点上下文中设置业务级信息
*
*
* @param businessKey
*/
public static void trace(String businessKey) {
@ -25,6 +24,10 @@ public final class BusinessKeyAppender {
if (spanData == null) {
return;
}
spanData.setBusinessKey(businessKey.replace('-', spiltChar).substring(0, Config.BusinessKey.MAX_LENGTH));
if (businessKey.length() <= Config.BusinessKey.MAX_LENGTH) {
spanData.setBusinessKey(businessKey);
return;
}
spanData.setBusinessKey(businessKey.substring(0, Config.BusinessKey.MAX_LENGTH));
}
}

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

@ -11,10 +11,9 @@ 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.*;
import static com.ai.cloud.skywalking.conf.Config.Sender.MAX_BUFFER_DATA_SIZE;
public class BufferGroup {
// private static Logger logger = Logger.getLogger(BufferGroup.class.getName());
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);
@ -72,26 +71,39 @@ public class BufferGroup {
continue;
}
bool = true;
data.append(dataBuffer[i] + ";");
dataBuffer[i] = null;
if (index++ == MAX_BUFFER_DATA_SIZE || data.length() >= Config.Sender.MAX_SEND_LENGTH) {
if (data.length() + dataBuffer[i].toString().length() >= Config.Sender.MAX_SEND_LENGTH) {
while (!DataSenderFactory.getSender().send(data.toString())) {
try {
Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL);
} catch (InterruptedException e) {
// logger.log(Level.ALL, "Sleep Failure");
logger.log(Level.ALL, "Sleep Failure");
}
}
index = 0;
data = new StringBuilder();
}
data.append(dataBuffer[i] + ";");
dataBuffer[i] = null;
}
if (data != null && data.length() > 0) {
while (!DataSenderFactory.getSender().send(data.toString())) {
try {
Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL);
} catch (InterruptedException e) {
logger.log(Level.ALL, "Sleep Failure");
}
}
index = 0;
data = new StringBuilder();
}
if (!bool) {
try {
Thread.sleep(MAX_WAIT_TIME);
} catch (InterruptedException e) {
// logger.log(Level.ALL, "Sleep Failure");
logger.log(Level.ALL, "Sleep Failure");
}
}
}

6
skywalking-api/src/main/java/com/ai/cloud/skywalking/constants/Constants.java Normal file
View File

@ -0,0 +1,6 @@
package com.ai.cloud.skywalking.constants;
public class Constants {
public static final String spiltRegx = "^|@|#";
}

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

@ -1,8 +1,10 @@
package com.ai.cloud.skywalking.context;
import com.ai.cloud.skywalking.constants.Constants;
import com.ai.cloud.skywalking.util.StringUtil;
public class Span {
private String traceId;
private String parentLevel;
private int levelId;
@ -132,53 +134,53 @@ public class Span {
@Override
public String toString() {
StringBuilder toStringValue = new StringBuilder();
toStringValue.append(traceId + "-");
toStringValue.append(traceId + Constants.spiltRegx);
if (!StringUtil.isEmpty(parentLevel)) {
toStringValue.append(parentLevel + "-");
toStringValue.append(parentLevel + Constants.spiltRegx);
} else {
toStringValue.append(" -");
toStringValue.append(" " + Constants.spiltRegx);
}
toStringValue.append(levelId + "-");
toStringValue.append(levelId + Constants.spiltRegx);
if (!StringUtil.isEmpty(viewPointId)) {
toStringValue.append(viewPointId + "-");
toStringValue.append(viewPointId + Constants.spiltRegx);
} else {
toStringValue.append(" -");
toStringValue.append(" " + Constants.spiltRegx);
}
toStringValue.append(startDate + "-");
toStringValue.append(cost + "-");
toStringValue.append(startDate + Constants.spiltRegx);
toStringValue.append(cost + Constants.spiltRegx);
if (!StringUtil.isEmpty(address)) {
toStringValue.append(address + "-");
toStringValue.append(address + Constants.spiltRegx);
} else {
toStringValue.append(" -");
toStringValue.append(" " + Constants.spiltRegx);
}
toStringValue.append(statusCode + "-");
toStringValue.append(statusCode + Constants.spiltRegx);
if (!StringUtil.isEmpty(exceptionStack)) {
toStringValue.append(exceptionStack + "-");
toStringValue.append(exceptionStack + Constants.spiltRegx);
} else {
toStringValue.append(" -");
toStringValue.append(" " + Constants.spiltRegx);
}
toStringValue.append(spanType + "-");
toStringValue.append(isReceiver + "-");
toStringValue.append(spanType + Constants.spiltRegx);
toStringValue.append(isReceiver + Constants.spiltRegx);
if (!StringUtil.isEmpty(businessKey)) {
toStringValue.append(businessKey + "-");
toStringValue.append(businessKey + Constants.spiltRegx);
} else {
toStringValue.append(" -");
toStringValue.append(" " + Constants.spiltRegx);
}
if (!StringUtil.isEmpty(processNo)) {
toStringValue.append(processNo);
} else {
toStringValue.append(" -");
toStringValue.append(" " + Constants.spiltRegx);
}
return toStringValue.toString();

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

@ -13,7 +13,7 @@ public final class ExceptionHandleUtil {
ByteArrayOutputStream buf = new ByteArrayOutputStream();
StringBuilder expMessage = new StringBuilder();
Throwable causeException = e;
while (causeException.getCause() != null || expMessage.length() < MAX_EXCEPTION_STACK_LENGTH) {
while (causeException != null && (causeException.getCause() != null || expMessage.length() < MAX_EXCEPTION_STACK_LENGTH)) {
causeException.printStackTrace(new java.io.PrintWriter(buf, true));
expMessage.append(buf.toString());
causeException = causeException.getCause();
@ -25,9 +25,9 @@ public final class ExceptionHandleUtil {
expMessage.append(e1.getCause().getMessage());
}
if (expMessage.length() <= MAX_EXCEPTION_STACK_LENGTH) {
return expMessage.toString().replace('\n', '&');
return expMessage.toString().replaceAll("\n", "&");
} else {
return expMessage.toString().replace('\n', '&').substring(0, MAX_EXCEPTION_STACK_LENGTH);
return expMessage.toString().replaceAll("\n", "&").substring(0, MAX_EXCEPTION_STACK_LENGTH);
}
}

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

@ -44,13 +44,15 @@ public class Config {
public static int OFFSET_FILE_SKIP_LENGTH = 2048;
// 每次读取文件偏移量大小
public static int OFFSET_FILE_READ_BUFFER_SIZE = 2048;
// 处理文件完成之后，等待时间
public static long SWITCH_FILE_WAIT_TIME = 5000L;
// 追加EOF标志位的线程数量
public static int MAX_APPEND_EOF_FLAGS_THREAD_NUMBER = 2;
// 每次存储的最大数量
public static final int MAX_STORAGE_SIZE_PER_TIME = 1024 * 1024;
}
public static class RegisterPersistence {

5
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/Constants.java Normal file
View File

@ -0,0 +1,5 @@
package com.ai.cloud.skywalking.reciever.constants;
public class Constants {
public static final String spiltRegx = "\\^\\|@\\|#";
}

45
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BuriedPointEntry.java
View File

@ -1,5 +1,7 @@
package com.ai.cloud.skywalking.reciever.model;
import com.ai.cloud.skywalking.reciever.constants.Constants;
import java.util.Date;
public class BuriedPointEntry {
@ -16,12 +18,17 @@ public class BuriedPointEntry {
private boolean isReceiver = false;
private String businessKey;
private String processNo;
// 元数据
private String originData;
private BuriedPointEntry() {
}
private BuriedPointEntry(String originData) {
}
public String getTraceId() {
return traceId;
}
@ -74,9 +81,13 @@ public class BuriedPointEntry {
return processNo;
}
public String getOriginData() {
return originData;
}
public static BuriedPointEntry convert(String str) {
BuriedPointEntry result = new BuriedPointEntry();
String[] fieldValues = str.split("-");
String[] fieldValues = str.split(Constants.spiltRegx);
result.traceId = fieldValues[0].trim();
result.parentLevel = fieldValues[1].trim();
result.levelId = Integer.valueOf(fieldValues[2]);
@ -84,12 +95,32 @@ public class BuriedPointEntry {
result.startDate = new Date(Long.valueOf(fieldValues[4]));
result.cost = Long.parseLong(fieldValues[5]);
result.address = fieldValues[6].trim();
result.exceptionStack = fieldValues[7].trim();
result.spanType = fieldValues[8].charAt(0);
result.isReceiver = Boolean.getBoolean(fieldValues[9]);
result.businessKey = fieldValues[10].replace('^', '-').trim();
result.processNo = fieldValues[11].trim();
result.statusCode = Byte.valueOf(fieldValues[7].trim());
result.exceptionStack = fieldValues[8].trim();
result.spanType = fieldValues[9].charAt(0);
result.isReceiver = Boolean.getBoolean(fieldValues[10]);
result.businessKey = fieldValues[11].trim();
result.processNo = fieldValues[12].trim();
result.originData = str;
return result;
}
@Override
public String toString() {
return "BuriedPointEntry{" +
"traceId='" + traceId + '\'' +
", parentLevel='" + parentLevel + '\'' +
", levelId=" + levelId +
", viewPointId='" + viewPointId + '\'' +
", startDate=" + startDate +
", cost=" + cost +
", address='" + address + '\'' +
", statusCode=" + statusCode +
", exceptionStack='" + exceptionStack + '\'' +
", spanType=" + spanType +
", isReceiver=" + isReceiver +
", businessKey='" + businessKey + '\'' +
", processNo='" + processNo + '\'' +
'}';
}
}

79
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java
View File

@ -1,30 +1,23 @@
package com.ai.cloud.skywalking.reciever.persistance;
import static com.ai.cloud.skywalking.reciever.conf.Config.Persistence.OFFSET_FILE_READ_BUFFER_SIZE;
import static com.ai.cloud.skywalking.reciever.conf.Config.Persistence.OFFSET_FILE_SKIP_LENGTH;
import static com.ai.cloud.skywalking.reciever.conf.Config.Persistence.SWITCH_FILE_WAIT_TIME;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.storage.StorageChainController;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.comparator.NameFileComparator;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.ai.cloud.skywalking.reciever.conf.Config;
import com.ai.cloud.skywalking.reciever.storage.StorageChainController;
import java.io.*;
import static com.ai.cloud.skywalking.reciever.conf.Config.Persistence.*;
public class PersistenceThread extends Thread {
private Logger logger = LogManager.getLogger(PersistenceThread.class);
@Override
public void run() {
int length;
File file1;
BufferedReader bufferedReader;
int offset;
@ -46,40 +39,48 @@ public class PersistenceThread extends Thread {
if (logger.isDebugEnabled()) {
logger.debug("Get file[{}] offset [{}]", file1.getName(), offset);
}
char[] chars = new char[OFFSET_FILE_READ_BUFFER_SIZE];
data = new StringBuffer(2048);
boolean bool = true;
length = 0;
while (bool) {
if ((length = bufferedReader.read(chars, 0, chars.length)) == -1) {
StringBuilder stringBuilder = new StringBuilder(MAX_STORAGE_SIZE_PER_TIME);
String tmpData;
while (true) {
tmpData = bufferedReader.readLine();
if (tmpData == null || tmpData.length() <= 0) {
if (stringBuilder != null && stringBuilder.length() > 0) {
StorageChainController.doStorage(stringBuilder.toString());
stringBuilder.delete(0, stringBuilder.length());
}
MemoryRegister.instance().doRegisterStatus(new FileRegisterEntry(file1.getName(), offset,
FileRegisterEntry.FileRegisterEntryStatus.UNREGISTER));
break;
}
offset += length;
for (int i = 0; i < chars.length; i++) {
if (chars[i] != '\n') {
data.append(chars[i]);
continue;
if ("EOF".equals(tmpData)) {
if (stringBuilder != null && stringBuilder.length() > 0) {
StorageChainController.doStorage(stringBuilder.toString());
}
if ("EOF".equals(data.toString())) {
bufferedReader.close();
logger.info("Data in file[{}] has been successfully processed", file1.getName());
boolean deleteSuccess = false;
while (!deleteSuccess) {
deleteSuccess = FileUtils.deleteQuietly(new File(file1.getParent(), file1.getName()));
}
logger.info("Delete file[{}] {}", file1.getName(), (deleteSuccess ? "success" : "failed"));
MemoryRegister.instance().unRegister(file1.getName());
bool = false;
break;
bufferedReader.close();
logger.info("Data in file[{}] has been successfully processed", file1.getName());
boolean deleteSuccess = false;
while (!deleteSuccess) {
deleteSuccess = FileUtils.deleteQuietly(new File(file1.getParent(), file1.getName()));
}
StorageChainController.doStorage(data.toString());
data.delete(0, data.length());
logger.info("Delete file[{}] {}", file1.getName(), (deleteSuccess ? "success" : "failed"));
MemoryRegister.instance().unRegister(file1.getName());
break;
}
if (stringBuilder.length() + tmpData.length() >= MAX_STORAGE_SIZE_PER_TIME) {
StorageChainController.doStorage(stringBuilder.toString());
stringBuilder.delete(0, stringBuilder.length());
MemoryRegister.instance().doRegisterStatus(new FileRegisterEntry(file1.getName(), offset,
FileRegisterEntry.FileRegisterEntryStatus.REGISTER));
}
stringBuilder.append(tmpData);
// 加上回车的字符串长度
offset += tmpData.length() + 1;
}
} catch (FileNotFoundException e) {
logger.error("The data file could not be found", e);

4
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/Chain.java
View File

@ -13,9 +13,9 @@ public class Chain {
this.chains = chains;
}
public void doChain(BuriedPointEntry entry, String entryOriginData ){
public void doChain(List<BuriedPointEntry> enties){
if(index < chains.size()){
chains.get(index++).doChain(entry, entryOriginData, this);;
chains.get(index++).doChain(enties,this);
}
}

4
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/IStorageChain.java
View File

@ -2,6 +2,8 @@ package com.ai.cloud.skywalking.reciever.storage;
import com.ai.cloud.skywalking.reciever.model.BuriedPointEntry;
import java.util.List;
public interface IStorageChain {
public void doChain(BuriedPointEntry entry, String entryOriginData, Chain chain);
void doChain(List<BuriedPointEntry> entry, Chain chain);
}

26
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/StorageChainController.java
View File

@ -24,25 +24,31 @@ public class StorageChainController {
if (buriedPointData == null || buriedPointData.length == 0) {
return;
}
List<BuriedPointEntry> entries = new ArrayList<BuriedPointEntry>();
for (String buriedPoint : buriedPointData) {
try {
if (buriedPoint == null || buriedPoint.trim().length() == 0) {
continue;
}
BuriedPointEntry entry = BuriedPointEntry.convert(buriedPoint);
while(true) {
try {
Chain chain = new Chain(chainArray);
chain.doChain(entry, buriedPoint);
break;
} catch (Throwable e) {
Thread.sleep(Config.StorageChain.RETRY_STORAGE_WAIT_TIME);
}
}
entries.add(BuriedPointEntry.convert(buriedPoint));
} catch (Throwable e) {
logger.error("ready to save buriedPoint error, choose to ignore. data="
+ buriedPoint, e);
}
}
while (true) {
try {
Chain chain = new Chain(chainArray);
chain.doChain(entries);
break;
} catch (Throwable e) {
try {
Thread.sleep(Config.StorageChain.RETRY_STORAGE_WAIT_TIME);
} catch (InterruptedException e1) {
logger.error("Sleep failure", e);
}
}
}
}
}

79
skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/SaveToHBaseChain.java
View File

@ -23,14 +23,9 @@ public class SaveToHBaseChain implements IStorageChain {
private static Connection connection;
@Override
public void doChain(BuriedPointEntry entry, String entryOriginData, Chain chain) {
if (StringUtils.isEmpty(entry.getParentLevel().trim())) {
insert(entry.getTraceId(), String.valueOf(entry.getLevelId()), entryOriginData);
} else {
insert(entry.getTraceId(), entry.getParentLevel() + "." + entry.getLevelId(), entryOriginData);
}
chain.doChain(entry, entryOriginData);
public void doChain(List<BuriedPointEntry> entry, Chain chain) {
insertBatch(entry);
chain.doChain(entry);
}
private static void initHBaseClient() throws IOException {
@ -81,4 +76,72 @@ public class SaveToHBaseChain implements IStorageChain {
}
}
public static boolean insert(String tableName, Put put) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
table.put(put);
if (logger.isDebugEnabled()) {
logger.debug("Insert data[RowKey:{}] success.", put.getId());
}
return true;
} catch (IOException e) {
logger.error("Insert the data error.RowKey:[{}]", put.getId(), e);
return false;
}
}
private static void insertBatch(List<BuriedPointEntry> entries) {
List<Put> puts = new ArrayList<Put>();
Put put;
for (BuriedPointEntry buriedPointEntry : entries) {
put = new Put(Bytes.toBytes(buriedPointEntry.getTraceId()));
if (StringUtils.isEmpty(buriedPointEntry.getParentLevel().trim())) {
put.addColumn(Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME), Bytes.toBytes(buriedPointEntry.getLevelId()),
Bytes.toBytes(buriedPointEntry.getOriginData()));
} else {
put.addColumn(Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME), Bytes.toBytes(buriedPointEntry.getParentLevel()
+ "." + buriedPointEntry.getLevelId()), Bytes.toBytes(buriedPointEntry.getOriginData()));
}
puts.add(put);
}
insertBatch(Config.HBaseConfig.TABLE_NAME, puts);
}
private static void insertBatch(String tableName, List<Put> data) {
Object[] resultArrays = new Object[data.size()];
try {
Table table = connection.getTable(TableName.valueOf(tableName));
table.batch(data, resultArrays);
int index = 0;
for (Object result : resultArrays) {
if (result == null) {
while (!insert(tableName, data.get(index))) {
Thread.sleep(100L);
}
}
index++;
}
} catch (IOException e) {
throw new ChainException(e);
} catch (InterruptedException e) {
throw new ChainException(e);
}
}
public static List<BuriedPointEntry> selectByTraceId(String traceId) throws IOException {
List<BuriedPointEntry> entries = new ArrayList<BuriedPointEntry>();
Table table = connection.getTable(TableName.valueOf(Config.HBaseConfig.TABLE_NAME));
Get g = new Get(Bytes.toBytes(traceId));
Result r = table.get(g);
for (Cell cell : r.rawCells()) {
if (cell.getValueArray().length > 0)
entries.add(BuriedPointEntry.convert(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())));
}
return entries;
}
}

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

@ -1,5 +1,5 @@
#采集服务器的端口
server.port=35000
server.port=34000
#最大数据缓存线程数量
buffer.max_thread_number=1