From b5761a9c385c96da709a57014590108c8092ec55 Mon Sep 17 00:00:00 2001 From: wusheng Date: Mon, 23 Nov 2015 23:01:34 +0800 Subject: [PATCH] =?UTF-8?q?1.=E5=8A=A0=E5=85=A5=E6=9C=8D=E5=8A=A1=E7=AB=AF?= =?UTF-8?q?=E7=9B=91=E6=8E=A7=E6=95=B0=E6=8D=AE=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cloud/skywalking/buffer/BufferGroup.java | 2 +- .../reciever/buffer/DataBufferThread.java | 17 +- .../skywalking/reciever/conf/Constants.java | 7 + .../reciever/constants/Constants.java | 5 - .../handler/CollectionServerDataHandler.java | 5 +- .../reciever/model/BuriedPointEntry.java | 2 +- .../persistance/PersistenceThread.java | 272 ++++++++++-------- .../RegisterPersistenceThread.java | 10 +- .../ServerHealthCollector.java | 33 +++ .../selfexamination/ServerHeathReading.java | 74 +++++ .../storage/chain/SaveToHBaseChain.java | 27 +- .../skywalking/reciever/util/MachineUtil.java | 64 +++++ .../skywalking/reciever/util/StringUtil.java | 10 + 13 files changed, 376 insertions(+), 152 deletions(-) create mode 100644 skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Constants.java delete mode 100644 skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/Constants.java create mode 100644 skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHealthCollector.java create mode 100644 skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHeathReading.java create mode 100644 skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/MachineUtil.java create mode 100644 skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/StringUtil.java 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 c849e12d5..31fcb8090 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 @@ -39,7 +39,7 @@ public class BufferGroup { public void save(Span span) { int i = Math.abs(index.getAndIncrement() % BUFFER_MAX_SIZE); if (dataBuffer[i] != null) { - HealthCollector.getCurrentHeathReading(null).updateData(HeathReading.WARNING, span.getLevelId() + "在Group[" + groupName + "]的第" + i + "位冲突"); + HealthCollector.getCurrentHeathReading(null).updateData(HeathReading.WARNING, "Group[" + groupName + "] index[" + i + "] data collision, discard old data."); } dataBuffer[i] = span; } diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java index 9811d3090..a599ea953 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java @@ -3,6 +3,9 @@ package com.ai.cloud.skywalking.reciever.buffer; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector; +import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading; + import java.io.File; import java.io.FileNotFoundException; import java.io.FileOutputStream; @@ -21,6 +24,7 @@ public class DataBufferThread extends Thread { private AtomicInteger index = new AtomicInteger(); public DataBufferThread() { + super("DataBufferThread"); try { file = new File(DATA_BUFFER_FILE_PARENT_DIRECTORY, getFileName()); if (file.exists()) { @@ -69,6 +73,7 @@ public class DataBufferThread extends Thread { if (index++ > FLUSH_NUMBER_OF_CACHE) { try { outputStream.flush(); + ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer flush data to local file:" + file.getName()); } catch (IOException e) { logger.error("Flush buffer data failed.", e); } finally { @@ -82,13 +87,14 @@ public class DataBufferThread extends Thread { if (hasData2Flush){ try { outputStream.flush(); + ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer flush data to local file:" + file.getName()); } catch (IOException e) { logger.error("Flush buffer data failed.", e); } } if (file.length() > DATA_FILE_MAX_LENGTH) { - convertFile(); + switchFile(); } if (!hasData2Flush) { @@ -106,7 +112,7 @@ public class DataBufferThread extends Thread { return System.currentTimeMillis() + "-" + UUID.randomUUID().toString().replaceAll("-", ""); } - private void convertFile() { + private void switchFile() { String fileName = getFileName(); try { @@ -120,27 +126,32 @@ public class DataBufferThread extends Thread { } catch (IOException e) { logger.error("close cache data failed.", e); } + ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer close local file:" + file.getName()); } logger.debug("Begin to switch the data file to {}.", fileName); try { file = new File(DATA_BUFFER_FILE_PARENT_DIRECTORY, fileName); outputStream = new FileOutputStream(file, true); + ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer open new local file:" + file.getName()); } catch (IOException e) { + ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.ERROR, "DataBuffer open new local file failure."); logger.error("Switch data file failed.", e); } } - public void doCarry(byte[] s) { + public void saveTemporarily(byte[] s) { int i = Math.abs(index.getAndIncrement() % data.length); while (data[i] != null) { try { + ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.WARNING, "DataBuffer index[" + i + "] data collision, service pausing. "); Thread.sleep(DATA_CONFLICT_WAIT_TIME); } catch (InterruptedException e) { logger.error("Failure sleep.", e); } } + ServerHealthCollector.getCurrentHeathReading(null).updateData(ServerHeathReading.INFO, "DataBuffer reveiving data."); data[i] = s; } diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Constants.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Constants.java new file mode 100644 index 000000000..318b23178 --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Constants.java @@ -0,0 +1,7 @@ +package com.ai.cloud.skywalking.reciever.conf; + +public class Constants { + public static final String spiltRegx = "\\^\\~"; + + public static final String HEALTH_DATA_SPILT_PATTERN = "^~"; +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/Constants.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/Constants.java deleted file mode 100644 index 2ae6311d9..000000000 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/Constants.java +++ /dev/null @@ -1,5 +0,0 @@ -package com.ai.cloud.skywalking.reciever.constants; - -public class Constants { - public static final String spiltRegx = "\\^\\~"; -} 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 bd2f6ec0e..30b122e23 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 @@ -6,8 +6,11 @@ import io.netty.channel.SimpleChannelInboundHandler; import com.ai.cloud.skywalking.reciever.buffer.DataBufferThreadContainer; public class CollectionServerDataHandler extends SimpleChannelInboundHandler { + @Override protected void channelRead0(ChannelHandlerContext ctx, byte[] msg) throws Exception { - DataBufferThreadContainer.getDataBufferThread().doCarry(msg); + Thread.currentThread().setName("ServerReceiver"); + + DataBufferThreadContainer.getDataBufferThread().saveTemporarily(msg); } } diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BuriedPointEntry.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BuriedPointEntry.java index 17621bfd8..4c849f785 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BuriedPointEntry.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/model/BuriedPointEntry.java @@ -1,6 +1,6 @@ package com.ai.cloud.skywalking.reciever.model; -import com.ai.cloud.skywalking.reciever.constants.Constants; +import com.ai.cloud.skywalking.reciever.conf.Constants; import java.util.ArrayList; import java.util.Date; diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java index 1075b2789..68550d6c3 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java @@ -1,7 +1,10 @@ package com.ai.cloud.skywalking.reciever.persistance; import com.ai.cloud.skywalking.reciever.conf.Config; +import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector; +import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading; 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; @@ -12,130 +15,171 @@ 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); + PersistenceThread() { + super("PersistenceThread"); + } - private Logger logger = LogManager.getLogger(PersistenceThread.class); + @Override + public void run() { + File file1; + BufferedReader bufferedReader = null; + int offset; + while (true) { + file1 = getDataFiles(); + if (file1 == null) { + try { + Thread.sleep(SWITCH_FILE_WAIT_TIME); + } catch (InterruptedException e) { + logger.error("Failure sleep", e); + } + continue; + } - @Override - public void run() { - File file1; - BufferedReader bufferedReader; - int offset; - while (true) { - file1 = getDataFiles(); - if (file1 == null) { - try { - Thread.sleep(SWITCH_FILE_WAIT_TIME); - } catch (InterruptedException e) { - logger.error("Failure sleep", e); - } - continue; - } + try { + bufferedReader = new BufferedReader(new FileReader(file1)); + offset = moveOffSet(file1, bufferedReader); + if (logger.isDebugEnabled()) { + logger.debug("Get file[{}] offset [{}]", file1.getName(), + offset); + } + StringBuilder stringBuilder = new StringBuilder( + MAX_STORAGE_SIZE_PER_TIME); + String tmpData; + while (true) { + tmpData = bufferedReader.readLine(); - try { - bufferedReader = new BufferedReader(new FileReader(file1)); - offset = moveOffSet(file1, bufferedReader); - if (logger.isDebugEnabled()) { - logger.debug("Get file[{}] offset [{}]", file1.getName(), offset); - } - 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; + } + ServerHealthCollector.getCurrentHeathReading(null) + .updateData( + ServerHeathReading.INFO, + "read " + tmpData.length() + + " chars from local file:" + file1.getName()); - 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; - } + if ("EOF".equals(tmpData)) { + if (stringBuilder != null && stringBuilder.length() > 0) { + StorageChainController.doStorage(stringBuilder + .toString()); + } - if ("EOF".equals(tmpData)) { + 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()); + break; + } - if (stringBuilder != null && stringBuilder.length() > 0) { - StorageChainController.doStorage(stringBuilder.toString()); - } + 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)); + } - 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()); - break; - } + stringBuilder.append(tmpData); + // 加上回车的字符串长度 + offset += tmpData.length() + 1; + } + } catch (FileNotFoundException e) { + logger.error("The data file could not be found", e); + } catch (IOException e) { + logger.error("The data file could not be found", e); + } finally { + try { + if (bufferedReader != null) + bufferedReader.close(); + } catch (IOException e) { + logger.error("can't close data file", e); + } + } - 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)); - } + try { + Thread.sleep(SWITCH_FILE_WAIT_TIME); + } catch (InterruptedException e) { + logger.error("Failure sleep.", e); + } + } + } - stringBuilder.append(tmpData); - // 加上回车的字符串长度 - offset += tmpData.length() + 1; - } - } catch (FileNotFoundException e) { - logger.error("The data file could not be found", e); - } catch (IOException e) { - logger.error("The data file could not be found", e); - } + private int moveOffSet(File file1, BufferedReader bufferedReader) + throws IOException { + int offset = MemoryRegister.instance().getOffSet(file1.getName()); + if (-1 == offset || offset == 0) { + // 以前该文件没有被任何人处理过,需要重新注册 + MemoryRegister + .instance() + .doRegisterStatus( + new FileRegisterEntry( + file1.getName(), + 0, + FileRegisterEntry.FileRegisterEntryStatus.REGISTER)); + offset = 0; + } else { + char[] cha = new char[OFFSET_FILE_SKIP_LENGTH]; + int length = 0; + while (length + OFFSET_FILE_SKIP_LENGTH < offset) { + length += OFFSET_FILE_SKIP_LENGTH; + bufferedReader.read(cha); + } + bufferedReader.read(cha, 0, Math.abs(offset - length)); + cha = null; + } + return offset; + } - try { - Thread.sleep(SWITCH_FILE_WAIT_TIME); - } catch (InterruptedException e) { - logger.error("Failure sleep.", e); - } - } - } + private File getDataFiles() { + File file1 = null; + File parentDir = new File( + Config.Buffer.DATA_BUFFER_FILE_PARENT_DIRECTORY); + NameFileComparator sizeComparator = new NameFileComparator(); + File[] dataFileList = sizeComparator.sort(parentDir.listFiles()); + for (File file : dataFileList) { + if (file.getName().startsWith(".")) { + continue; + } + if (MemoryRegister.instance().isRegister(file.getName())) { + if (logger.isDebugEnabled()) + logger.debug( + "The file [{}] is being used by another thread ", + file); + continue; + } + if (logger.isDebugEnabled()) { + logger.debug("Begin to deal data file [{}]", file.getName()); + } + file1 = file; + break; + } - private int moveOffSet(File file1, BufferedReader bufferedReader) throws IOException { - int offset = MemoryRegister.instance().getOffSet(file1.getName()); - if (-1 == offset || offset == 0) { - // 以前该文件没有被任何人处理过,需要重新注册 - MemoryRegister.instance().doRegisterStatus(new FileRegisterEntry(file1.getName(), 0, - FileRegisterEntry.FileRegisterEntryStatus.REGISTER)); - offset = 0; - } else { - char[] cha = new char[OFFSET_FILE_SKIP_LENGTH]; - int length = 0; - while (length + OFFSET_FILE_SKIP_LENGTH < offset) { - length += OFFSET_FILE_SKIP_LENGTH; - bufferedReader.read(cha); - } - bufferedReader.read(cha, 0, Math.abs(offset - length)); - cha = null; - } - return offset; - } - - private File getDataFiles() { - File file1 = null; - File parentDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIRECTORY); - NameFileComparator sizeComparator = new NameFileComparator(); - File[] dataFileList = sizeComparator.sort(parentDir.listFiles()); - for (File file : dataFileList) { - if (file.getName().startsWith(".")) { - continue; - } - if (MemoryRegister.instance().isRegister(file.getName())) { - if (logger.isDebugEnabled()) - logger.debug("The file [{}] is being used by another thread ", file); - continue; - } - if (logger.isDebugEnabled()) { - logger.debug("Begin to deal data file [{}]", file.getName()); - } - file1 = file; - break; - } - - return file1; - } + return file1; + } } diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/RegisterPersistenceThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/RegisterPersistenceThread.java index 9c0cd46c7..e238ec800 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/RegisterPersistenceThread.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/RegisterPersistenceThread.java @@ -19,7 +19,7 @@ public class RegisterPersistenceThread extends Thread { private BufferedWriter writer; public RegisterPersistenceThread() { - + super("RegisterPersistenceThread"); } @Override @@ -30,7 +30,7 @@ public class RegisterPersistenceThread extends Thread { } catch (InterruptedException e) { logger.error("Sleep failure", e); } - // + File file = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_FILE_NAME); File bakFile = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_BAK_FILE_NAME); try { @@ -38,15 +38,15 @@ public class RegisterPersistenceThread extends Thread { } catch (IOException e) { logger.error("Sleep failure", e); } + Collection fileRegisterEntries = MemoryRegister.instance().getEntries(); - logger.info("file Register Entries size [{}]", fileRegisterEntries.size()); + logger.debug("file Register Entries size [{}]", fileRegisterEntries.size()); try { writer = new BufferedWriter(new FileWriter(file)); } catch (IOException e) { logger.error("Write The offset file anomalies."); - System.exit(-1); } - // + for (FileRegisterEntry fileRegisterEntry : fileRegisterEntries) { try { writer.write(fileRegisterEntry.toString() + "\n"); diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHealthCollector.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHealthCollector.java new file mode 100644 index 000000000..d100ba6d9 --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHealthCollector.java @@ -0,0 +1,33 @@ +package com.ai.cloud.skywalking.reciever.selfexamination; + +import java.util.HashMap; +import java.util.Map; + +import com.ai.cloud.skywalking.reciever.util.MachineUtil; + +public class ServerHealthCollector extends Thread{ + private static Map heathReadings = new HashMap(); + + public static ServerHeathReading getCurrentHeathReading(String extraId){ + String id = getId(extraId); + if(!heathReadings.containsKey(id)){ + synchronized (heathReadings) { + if(!heathReadings.containsKey(id)){ + heathReadings.put(id, new ServerHeathReading(id)); + } + } + } + return heathReadings.get(id); + } + + private static String getId(String extraId){ + return "SkyWalkingServer,M:" + MachineUtil.getHostDesc() +",P:" + MachineUtil.getProcessNo() + ",T:" + + Thread.currentThread().getName() + "(" + + Thread.currentThread().getId() + ")" + (extraId == null? "" : ",extra:" + extraId); + } + + @Override + public void run(){ + //TODO: 服务端本地存储，用于将信息存储如数据库，并供前台展现，完成定时刷新 + } +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHeathReading.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHeathReading.java new file mode 100644 index 000000000..9dfcdfb9c --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/selfexamination/ServerHeathReading.java @@ -0,0 +1,74 @@ +package com.ai.cloud.skywalking.reciever.selfexamination; + +import java.util.HashMap; +import java.util.Map; + +import com.ai.cloud.skywalking.reciever.conf.Constants; + +public class ServerHeathReading { + 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 ServerHeathReading(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-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/SaveToHBaseChain.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/SaveToHBaseChain.java index 1232cea27..0cdbb113c 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/SaveToHBaseChain.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/storage/chain/SaveToHBaseChain.java @@ -2,9 +2,12 @@ package com.ai.cloud.skywalking.reciever.storage.chain; import com.ai.cloud.skywalking.reciever.conf.Config; import com.ai.cloud.skywalking.reciever.model.BuriedPointEntry; +import com.ai.cloud.skywalking.reciever.selfexamination.ServerHealthCollector; +import com.ai.cloud.skywalking.reciever.selfexamination.ServerHeathReading; import com.ai.cloud.skywalking.reciever.storage.Chain; import com.ai.cloud.skywalking.reciever.storage.ChainException; import com.ai.cloud.skywalking.reciever.storage.IStorageChain; + import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.*; @@ -56,27 +59,6 @@ public class SaveToHBaseChain implements IStorageChain { } } - public static void insert(String rowKey, String qualifier, String value) { - insert(Config.HBaseConfig.TABLE_NAME, rowKey, qualifier, value); - } - - public static void insert(String tableName, String rowKey, String qualifier, String value) { - try { - Table table = connection.getTable(TableName.valueOf(tableName)); - Put put = new Put(Bytes.toBytes(rowKey)); - put.addColumn(Bytes.toBytes(Config.HBaseConfig.FAMILY_COLUMN_NAME), Bytes.toBytes(qualifier), Bytes - .toBytes(value)); - table.put(put); - if (logger.isDebugEnabled()) { - logger.debug("Insert data[RowKey:{}] success.", rowKey); - } - } catch (IOException e) { - logger.error("Insert the data error.RowKey:[{}],Qualifier[{}],value[{}]", rowKey, qualifier, value, e); - throw new ChainException(e); - } - } - - public static boolean insert(String tableName, Put put) { try { Table table = connection.getTable(TableName.valueOf(tableName)); @@ -92,7 +74,6 @@ public class SaveToHBaseChain implements IStorageChain { } - private static void bulkInsertBuriedPointData(List entries) { if (entries == null || entries.size() <= 0) return; @@ -120,6 +101,8 @@ public class SaveToHBaseChain implements IStorageChain { } bulkInsertBuriedPointData(Config.HBaseConfig.TABLE_NAME, puts); + + ServerHealthCollector.getCurrentHeathReading("hbase").updateData(ServerHeathReading.INFO, "save " + entries.size() + " BuriedPointEntries." ); } private static void bulkInsertBuriedPointData(String tableName, List data) { diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/MachineUtil.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/MachineUtil.java new file mode 100644 index 000000000..82ceaaafa --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/MachineUtil.java @@ -0,0 +1,64 @@ +package com.ai.cloud.skywalking.reciever.util; + +import java.lang.management.ManagementFactory; +import java.net.InetAddress; +import java.net.UnknownHostException; + +public class MachineUtil { + private static String processNo; + private static String IP; + private static String hostName; + + static { + processNo = getProcessNo(); + } + + public static String getProcessNo() { + if (StringUtil.isEmpty(processNo)) { + String name = ManagementFactory.getRuntimeMXBean().getName(); + processNo = name.split("@")[0]; + } + return processNo; + } + + private static InetAddress getInetAddress() { + try { + return InetAddress.getLocalHost(); + } catch (UnknownHostException e) { + hostName = "unknown host!"; + } + return null; + + } + + public static String getHostIp() { + if (StringUtil.isEmpty(IP)) { + InetAddress netAddress = getInetAddress(); + if (null == netAddress) { + return null; + } + IP = netAddress.getHostAddress(); //get the ip address + } + return IP; + } + + public static String getHostName() { + if (StringUtil.isEmpty(hostName)) { + InetAddress netAddress = getInetAddress(); + if (null == netAddress) { + return null; + } + hostName = netAddress.getHostName(); //get the host address + } + return hostName; + } + + public static String getHostDesc(){ + return getHostName() + "/" + getHostIp(); + } + + private MachineUtil() { + // Non + } + +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/StringUtil.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/StringUtil.java new file mode 100644 index 000000000..2b54a0bd9 --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/StringUtil.java @@ -0,0 +1,10 @@ +package com.ai.cloud.skywalking.reciever.util; + +public final class StringUtil { + public static boolean isEmpty(String str) { + if (str == null || "".equals(str) || str.length() == 0) { + return true; + } + return false; + } +}