diff --git a/skywalking-server/installer/bin/swEnv.sh b/skywalking-server/installer/bin/swEnv.sh new file mode 100644 index 000000000..d4154ee74 --- /dev/null +++ b/skywalking-server/installer/bin/swEnv.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env bash + +SW_PREFIX="${SW_SERVER_BIN}/.." +SW_LOG_DIR="${SW_SERVER_BIN}/../log" +SW_CFG_DIR="${SW_SERVER_BIN}/../config" + +#设置Java home +if [ "$JAVA_HOME" != "" ]; then + JAVA="$JAVA_HOME/bin/java" +else + JAVA=java +fi + +CLASSPATH="$SW_CFG_DIR:$CLASSPATH" + +for i in "${SW_SERVER_BIN}"/lib/*.jar +do + CLASSPATH="$i:$CLASSPATH" +done + +if $cygwin +then + CLASSPATH=`cygpath -wp "$CLASSPATH"` +fi + +echo "CLASSPATH=$CLASSPATH" \ No newline at end of file diff --git a/skywalking-server/installer/bin/swServer.sh b/skywalking-server/installer/bin/swServer.sh new file mode 100644 index 000000000..75a066997 --- /dev/null +++ b/skywalking-server/installer/bin/swServer.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash + +# use POSTIX interface, symlink is followed automatically +SW_SERVER_BIN="${BASH_SOURCE-$0}" +SW_SERVER_BIN="$(dirname "${SW_SERVER_BIN}")" +SW_SERVER_BIN="$(cd "${SW_SERVER_BIN}"; pwd)" + +# 设置Skywalking的目录基本信息 +. "SW_SERVER_BIN/swEnv.sh" + + diff --git a/skywalking-server/installer/config/config.properties b/skywalking-server/installer/config/config.properties new file mode 100644 index 000000000..c21ff8088 --- /dev/null +++ b/skywalking-server/installer/config/config.properties @@ -0,0 +1,35 @@ +#采集服务器的端口 +server.port=34000 + +#最大数据缓存线程数量 +buffer.max_thread_number=1 +#每个线程最大缓存数量 +buffer.per_thread_max_buffer_number=1024 +#无数据处理时轮询等待时间(单位:毫秒) +buffer.max_wait_time=5000 +#数据冲突时等待时间(单位:毫秒) +buffer.data_conflict_wait_time=10 +#数据缓存文件目录 +buffer.data_buffer_file_parent_directory=D:/test-data/data/buffer +#缓存数据文件最大长度(单位:byte) +buffer.data_file_max_length=10240 +#每次Flush的缓存数据的个数 +buffer.flush_number_of_cache=30 + +#最大持久化的线程数量 +persistence.max_thread_number=1 +#定位文件时,每次读取偏移量跳过大小 +persistence.offset_file_skip_length=2048 +#每次读取文件偏移量大小 +persistence.offset_file_read_buffer_size=2048 +#处理文件完成之后，等待时间(单位:毫秒) +persistence.switch_file_wait_time=5000 + +#偏移量注册文件的目录 +registerpersistence.register_file_parent_directory=D:/test-data/data/offset +#偏移量注册文件名 +registerpersistence.register_file_name=offset.txt +#偏移量注册备份文件名 +registerpersistence.register_bak_file_name=offset.txt.bak +#偏移量写入文件等待周期(单位:毫秒) +registerpersistence.offset_written_file_wait_cycle=5000 \ No newline at end of file diff --git a/skywalking-server/installer/config/log4j2.xml b/skywalking-server/installer/config/log4j2.xml new file mode 100644 index 000000000..025d5b2fb --- /dev/null +++ b/skywalking-server/installer/config/log4j2.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/skywalking-server/pom.xml b/skywalking-server/pom.xml new file mode 100644 index 000000000..a40294c88 --- /dev/null +++ b/skywalking-server/pom.xml @@ -0,0 +1,79 @@ + + 4.0.0 + + com.ai.cloud + skywalking-server + 1.0-SNAPSHOT + jar + + skywalking-server + http://maven.apache.org + + + UTF-8 + + + + + junit + junit + 4.12 + test + + + org.apache.logging.log4j + log4j-core + 2.4.1 + + + commons-io + commons-io + 2.4 + + + + + + org.apache.maven.plugins + maven-resources-plugin + 2.6 + + + copy-resources + pre-integration-test + + copy-resources + + + ${project.basedir}/installer/config + + + src/main/resources + false + + + + + + + + maven-dependency-plugin + + + copy-dependencies + pre-integration-test + + copy-dependencies + + + + + ${project.basedir}/installer/lib + false + true + + + + + diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java new file mode 100644 index 000000000..9486a3f9d --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java @@ -0,0 +1,102 @@ +package com.ai.cloud.skywalking.reciever; + +import com.ai.cloud.skywalking.reciever.buffer.DataBufferThread; +import com.ai.cloud.skywalking.reciever.buffer.DataBufferThreadContainer; +import com.ai.cloud.skywalking.reciever.conf.Config; +import com.ai.cloud.skywalking.reciever.conf.ConfigInitializer; +import com.ai.cloud.skywalking.reciever.persistance.PersistenceThreadLauncher; +import com.ai.cloud.skywalking.reciever.util.ByteArrayUtil; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.*; +import java.util.Iterator; +import java.util.Properties; + +public class CollectionServer { + + static Logger logger = LogManager.getLogger(CollectionServer.class); + private Selector selector; + + public CollectionServer() { + } + + public void doCollect() throws IOException { + ServerSocketChannel serverSocketChannel = initServerSocketChannel(); + DataBufferThread dataBuffer; + while (selector.select() > 0) { + Iterator iterator = selector.selectedKeys().iterator(); + while (iterator.hasNext()) { + SelectionKey key = (SelectionKey) iterator.next(); + iterator.remove(); + beginToRead(serverSocketChannel, key); + if (key.isReadable()) { + ByteChannel sc = (SocketChannel) key.channel(); + ByteBuffer contextLengthBuffer = ByteBuffer.allocate(4); + try { + sc.read(contextLengthBuffer); + int length = ByteArrayUtil.byteArrayToInt(contextLengthBuffer.array(), 0); + if (length > 0) { + ByteBuffer contentBuffer = ByteBuffer.allocate(length); + sc.read(contentBuffer); + dataBuffer = DataBufferThreadContainer.getDataBufferThread(); + dataBuffer.doCarry(new String(contentBuffer.array())); + } + contextLengthBuffer.flip(); + } catch (IOException e) { + logger.error("The remote client disconnect service", e); + sc.close(); + } + } + } + } + } + + private void beginToRead(ServerSocketChannel serverSocketChannel, SelectionKey key) throws IOException { + if (key.isAcceptable()) { + SocketChannel sc = serverSocketChannel.accept(); + sc.configureBlocking(false); + sc.register(selector, SelectionKey.OP_READ); + } + } + + private ServerSocketChannel initServerSocketChannel() throws IOException { + selector = Selector.open(); + ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); + serverSocketChannel.bind(new InetSocketAddress("0.0.0.0", Config.Server.PORT)); + serverSocketChannel.configureBlocking(false); + serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); + logger.info("The service is listening on port {}", Config.Server.PORT); + return serverSocketChannel; + } + + + public static void main(String[] args) throws IOException, IllegalAccessException { + logger.info("To initialize the collect server configuration parameters...."); + initializeParam(); + logger.info("To launch register persistence thread...."); + PersistenceThreadLauncher.doLaunch(); + logger.info("To init data buffer thread container..."); + DataBufferThreadContainer.init(); + logger.info("Starting collection server....."); + new CollectionServer().doCollect(); + } + + private static void initializeParam() throws IllegalAccessException, IOException { + Properties properties = new Properties(); + try { + properties.load(CollectionServer.class.getResourceAsStream("/config.properties")); + ConfigInitializer.initialize(properties, Config.class); + } catch (IllegalAccessException e) { + logger.error("Initialize the collect server configuration failed", e); + throw e; + } catch (IOException e) { + logger.error("Initialize the collect server configuration failed", e); + throw e; + } + } + +} 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 new file mode 100644 index 000000000..27ec6f84e --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java @@ -0,0 +1,132 @@ +package com.ai.cloud.skywalking.reciever.buffer; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.*; + +public class DataBufferThread extends Thread { + + private Logger logger = LogManager.getLogger(DataBufferThread.class); + private String[] data = new String[PER_THREAD_MAX_BUFFER_NUMBER]; + private File file; + private FileOutputStream outputStream; + private AtomicInteger index = new AtomicInteger(); + + public DataBufferThread() { + try { + file = new File(DATA_BUFFER_FILE_PARENT_DIRECTORY, getFileName()); + if (file.exists()) { + file.createNewFile(); + } + + if (logger.isDebugEnabled()){ + logger.debug("Create buffer data file {}.", file.getName()); + } + outputStream = new FileOutputStream(file, true); + } catch (FileNotFoundException e) { + logger.error("Data cache file cannot be created or written, please check the file system.", e); + System.exit(-1); + } catch (IOException e) { + logger.error("Data cache file cannot be created or written, please check the file system.", e); + System.exit(-1); + } + } + + @Override + public void run() { + boolean isWriteFailure; + while (true) { + boolean bool = false; + int count = 0; + int index = 0; + for (int i = 0; i < data.length; i++, count++) { + if (data[i] == null) { + continue; + } + bool = true; + isWriteFailure = true; + while (isWriteFailure) { + try { + outputStream.write((data[i] + "\n").getBytes()); + isWriteFailure = false; + } catch (IOException e) { + logger.error("Write buffer data failed.", e); + try { + Thread.sleep(WRITE_DATA_FAILURE_RETRY_INTERVAL); + } catch (InterruptedException e1) { + logger.error("Failure sleep.", e); + } + } + } + if (index++ > FLUSH_NUMBER_OF_CACHE) { + try { + outputStream.flush(); + } catch (IOException e) { + logger.error("Flush buffer data failed.", e); + } + } + data[i] = null; + + } + + if (file.length() > DATA_FILE_MAX_LENGTH) { + convertFile(); + } + + if (!bool) { + try { + Thread.sleep(MAX_WAIT_TIME); + } catch (InterruptedException e) { + logger.error("Failure sleep.", e); + } + } + } + } + + + private String getFileName() { + return System.currentTimeMillis() + "-" + UUID.randomUUID().toString().replaceAll("-", ""); + } + + private void convertFile() { + String fileName = getFileName(); + + try { + outputStream.write("EOF\n".getBytes()); + outputStream.flush(); + outputStream.close(); + } catch (IOException e) { + logger.error("Write cache data failed.", e); + } + 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); + } catch (IOException e) { + logger.error("Switch data file failed.", e); + } + + + } + + public void doCarry(String s) { + int i = Math.abs(index.getAndIncrement() % data.length); + while (data[i] != null) { + try { + Thread.sleep(DATA_CONFLICT_WAIT_TIME); + } catch (InterruptedException e) { + logger.error("Failure sleep.", e); + } + } + + data[i] = s; + } +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java new file mode 100644 index 000000000..2fde17fd1 --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThreadContainer.java @@ -0,0 +1,58 @@ +package com.ai.cloud.skywalking.reciever.buffer; + +import com.ai.cloud.skywalking.reciever.conf.Config; +import org.apache.commons.io.comparator.NameFileComparator; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ThreadLocalRandom; + +import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.MAX_THREAD_NUMBER; + +public class DataBufferThreadContainer { + private static Logger logger = LogManager.getLogger(DataBufferThreadContainer.class); + private static List buffers = new ArrayList(); + + private DataBufferThreadContainer() { + } + + public static DataBufferThread getDataBufferThread() { + if (buffers.size() == 0) { + throw new RuntimeException("Data buffer thread pool is not init"); + } + return buffers.get(ThreadLocalRandom.current().nextInt(buffers.size())); + } + + public static void init() { + logger.info("Add EOF flags to the unprocessed data file last time."); + File parentDir = new File(Config.Buffer.DATA_BUFFER_FILE_PARENT_DIRECTORY); + NameFileComparator sizeComparator = new NameFileComparator(); + File[] dataFileList = sizeComparator.sort(parentDir.listFiles()); + BufferedWriter bufferedWriter; + for (File file : dataFileList) { + try { + logger.info("Add EOF flags to unprocessed data file[{}]", file.getName()); + bufferedWriter = new BufferedWriter(new FileWriter(new File(file.getParent(), file.getName()), true)); + bufferedWriter.write("EOF\n"); + bufferedWriter.flush(); + bufferedWriter.close(); + } catch (IOException e) { + logger.info("Add EOF flags to the unprocessed data file failed.", e); + } + } + + logger.info("Data buffer thread size {} begin to init ", MAX_THREAD_NUMBER); + for (int i = 0; i < MAX_THREAD_NUMBER; i++) { + DataBufferThread dataBufferThread = new DataBufferThread(); + dataBufferThread.start(); + buffers.add(dataBufferThread); + } + } + +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java new file mode 100644 index 000000000..8c33e9d4d --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java @@ -0,0 +1,68 @@ +package com.ai.cloud.skywalking.reciever.conf; + +public class Config { + + // 采集服务配置类 + public static class Server { + // 采集服务器的端口 + public static int PORT = 34000; + } + + // 数据缓存配置类 + public static class Buffer { + // 最大数据缓存线程数量 + public static int MAX_THREAD_NUMBER = 1; + + //每个线程最大缓存数量 + public static int PER_THREAD_MAX_BUFFER_NUMBER = 1024; + + // 无数据处理时轮询等待时间(单位:毫秒) + public static long MAX_WAIT_TIME = 5000L; + + // 数据冲突时等待时间(单位:毫秒) + public static long DATA_CONFLICT_WAIT_TIME = 10L; + + // 数据缓存文件目录 + public static String DATA_BUFFER_FILE_PARENT_DIRECTORY = "../data/buffer"; + + // 缓存数据文件最大长度(单位:byte) + public static int DATA_FILE_MAX_LENGTH = 30 * 1024 * 1024; + + // 每次缓存数据写入失败，最大尝试时间 + public static long WRITE_DATA_FAILURE_RETRY_INTERVAL = 10 * 60 * 1000L; + + //每次Flush的缓存数据的个数 + public static int FLUSH_NUMBER_OF_CACHE = 30; + + } + + public static class Persistence { + // 最大持久化的线程数量 + public static int MAX_THREAD_NUMBER = 1; + + // 定位文件时,每次读取偏移量跳过大小 + 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; + } + + public static class RegisterPersistence { + // 偏移量注册文件的目录 + public static String REGISTER_FILE_PARENT_DIRECTORY = "../data/offset"; + + // 偏移量注册文件名 + public static String REGISTER_FILE_NAME = "offset.txt"; + + // 偏移量注册备份文件名 + public static String REGISTER_BAK_FILE_NAME = "offset.txt.bak"; + + // 偏移量写入文件等待周期 + public static long OFFSET_WRITTEN_FILE_WAIT_CYCLE = 5000L; + + } + +} \ No newline at end of file diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/ConfigInitializer.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/ConfigInitializer.java new file mode 100644 index 000000000..a729b5ce4 --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/ConfigInitializer.java @@ -0,0 +1,92 @@ +package com.ai.cloud.skywalking.reciever.conf; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.util.LinkedList; +import java.util.Properties; + +public class ConfigInitializer { + + public static Logger logger = LogManager.getLogger(ConfigInitializer.class); + + public static void initialize(Properties properties, Class rootConfigType) throws IllegalAccessException { + for (Class innerConfiguration : rootConfigType.getClasses()) { + for (Field field : innerConfiguration.getFields()) { + if (9 != field.getModifiers()) + continue; + String configKey = (innerConfiguration.getSimpleName() + "." + + field.getName()).toLowerCase(); + String value = properties.getProperty(configKey); + if (value != null) { + if (field.getType().equals(int.class)) + field.set(null, Integer.valueOf(value)); + if (field.getType().equals(String.class)) + field.set(null, value); + if (field.getType().equals(long.class)) + field.set(null, Long.valueOf(value)); + } + logger.debug("{}={}", configKey, value); + } + } + } + + public static void initialize2(Properties properties, Class rootConfigType) throws IllegalAccessException { + initNextLevel(properties, rootConfigType, new ConfigDesc()); + } + + private static void initNextLevel(Properties properties, Class recentConfigType, ConfigDesc parentDesc) throws NumberFormatException, IllegalArgumentException, IllegalAccessException { + for (Field field : recentConfigType.getFields()) { + if (Modifier.isPublic(field.getModifiers()) && Modifier.isStatic(field.getModifiers())) { + String configKey = (parentDesc + "." + + field.getName()).toLowerCase(); + String value = properties.getProperty(configKey); + if (value != null) { + if (field.getType().equals(int.class)) + field.set(null, Integer.valueOf(value)); + if (field.getType().equals(String.class)) + field.set(null, value); + if (field.getType().equals(long.class)) + field.set(null, Long.valueOf(value)); + } + logger.debug("{}={}", configKey, field.get(null)); + } + } + for (Class innerConfiguration : recentConfigType.getClasses()) { + parentDesc.append(innerConfiguration.getSimpleName()); + initNextLevel(properties, innerConfiguration, parentDesc); + parentDesc.removeLastDesc(); + } + } +} + +class ConfigDesc { + private LinkedList descs = new LinkedList(); + + void append(String currentDesc) { + descs.addLast(currentDesc); + } + + void removeLastDesc() { + descs.removeLast(); + } + + @Override + public String toString() { + if (descs.size() == 0) { + return ""; + } + StringBuilder ret = new StringBuilder(descs.getFirst()); + boolean first = true; + for (String desc : descs) { + if (first) { + first = false; + continue; + } + ret.append(".").append(desc); + } + return ret.toString(); + } +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/ServerConstants.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/ServerConstants.java new file mode 100644 index 000000000..26630292d --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/constants/ServerConstants.java @@ -0,0 +1,4 @@ +package com.ai.cloud.skywalking.reciever.constants; + +public class ServerConstants { +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/FileRegisterEntry.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/FileRegisterEntry.java new file mode 100644 index 000000000..1878eb940 --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/FileRegisterEntry.java @@ -0,0 +1,76 @@ +package com.ai.cloud.skywalking.reciever.persistance; + +public class FileRegisterEntry { + private String fileName; + private int offset; + private FileRegisterEntryStatus statue; + + public FileRegisterEntry() { + } + + public FileRegisterEntry(String fileName) { + this.fileName = fileName; + } + + public FileRegisterEntry(String fileName, int offset) { + this.fileName = fileName; + this.offset = offset; + } + + public FileRegisterEntry(String fileName, int offset, FileRegisterEntryStatus statue) { + this.fileName = fileName; + this.offset = offset; + this.statue = statue; + } + + public String getFileName() { + return fileName; + } + + public void setFileName(String fileName) { + this.fileName = fileName; + } + + public int getOffset() { + return offset; + } + + public void setOffset(int offset) { + this.offset = offset; + } + + public FileRegisterEntryStatus getStatue() { + return statue; + } + + public void setStatue(FileRegisterEntryStatus statue) { + this.statue = statue; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof FileRegisterEntry)) return false; + + FileRegisterEntry that = (FileRegisterEntry) o; + + return !(getFileName() != null ? !getFileName().equals(that.getFileName()) : that.getFileName() != null); + + } + + @Override + public int hashCode() { + return getFileName() != null ? getFileName().hashCode() : 0; + } + + + @Override + public String toString() { + return fileName + '\t' + offset + "\t" + statue; + } + + public enum FileRegisterEntryStatus { + REGISTER, UNREGISTER; + } + +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/MemoryRegister.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/MemoryRegister.java new file mode 100644 index 000000000..b2894926e --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/MemoryRegister.java @@ -0,0 +1,93 @@ +package com.ai.cloud.skywalking.reciever.persistance; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.*; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import static com.ai.cloud.skywalking.reciever.conf.Config.RegisterPersistence.REGISTER_FILE_NAME; +import static com.ai.cloud.skywalking.reciever.conf.Config.RegisterPersistence.REGISTER_FILE_PARENT_DIRECTORY; + +public class MemoryRegister { + private Logger logger = LogManager.getLogger(MemoryRegister.class); + private Map entries = new ConcurrentHashMap(); + private File file; + private static MemoryRegister memoryRegister = new MemoryRegister(); + + public static MemoryRegister instance() { + return memoryRegister; + } + + public synchronized void doRegisterStatus(FileRegisterEntry fileRegisterEntry) { + if (logger.isDebugEnabled()) { + logger.debug("Register entry[{}] into the memory register", fileRegisterEntry.getFileName()); + } + entries.put(fileRegisterEntry.getFileName(), fileRegisterEntry); + } + + public void unRegister(String fileName) { + if (logger.isDebugEnabled()) { + logger.debug("Unregister[{}] from the memory register", fileName); + } + entries.remove(fileName); + } + + public synchronized boolean isRegister(String fileName) { + if (entries.containsKey(fileName)) { + if (entries.get(fileName).getStatue() == FileRegisterEntry.FileRegisterEntryStatus.REGISTER) { + if (logger.isDebugEnabled()) { + logger.debug("Entry[{}] has been register", fileName); + } + return true; + } + } + return false; + } + + + public Collection getEntries() { + return entries.values(); + } + + public int getOffSet(String fileName) { + if (entries.containsKey(fileName)) { + return entries.get(fileName).getOffset(); + } + return -1; + } + + private void checkOffSetExists() { + file = new File(REGISTER_FILE_PARENT_DIRECTORY, REGISTER_FILE_NAME); + + if (!file.exists()) { + try { + file.createNewFile(); + } catch (IOException e) { + logger.error("Create offset filed failed", e); + } + } + } + + private MemoryRegister() { + // 读取offset文件 + checkOffSetExists(); + BufferedReader reader; + // 在处理数据之前需要初始化处理文件的处理状态 + try { + reader = new BufferedReader(new FileReader(file)); + String offsetData; + while ((offsetData = reader.readLine()) != null && !"EOF".equals(offsetData)) { + String[] ss = offsetData.split("\t"); + entries.put(ss[0], new FileRegisterEntry(ss[0], Integer.valueOf(ss[1]), FileRegisterEntry.FileRegisterEntryStatus.UNREGISTER)); + } + } catch (FileNotFoundException e) { + logger.error("The offset file does not exist.", e); + checkOffSetExists(); + } catch (IOException e) { + logger.error("Read data from offset file failed.", e); + } + } +} 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 new file mode 100644 index 000000000..b02ba6e60 --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java @@ -0,0 +1,125 @@ +package com.ai.cloud.skywalking.reciever.persistance; + +import com.ai.cloud.skywalking.reciever.conf.Config; +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 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() { + while (true) { + File file1 = getDataFiles(); + if (file1 == null) { + try { + Thread.sleep(SWITCH_FILE_WAIT_TIME); + } catch (InterruptedException e) { + logger.error("Failure sleep", e); + } + continue; + } + + try { + BufferedReader bufferedReader = new BufferedReader(new FileReader(file1)); + int offset = moveOffSet(file1, bufferedReader); + if (logger.isDebugEnabled()) { + logger.debug("Get file[{}] offset [{}]", file1.getName(), offset); + } + char[] chars = new char[OFFSET_FILE_READ_BUFFER_SIZE]; + StringBuffer data = new StringBuffer(); + boolean bool = true; + while (bool) { + if (bufferedReader.read(chars, 0, chars.length) == -1) { + MemoryRegister.instance().doRegisterStatus(new FileRegisterEntry(file1.getName(), offset, + FileRegisterEntry.FileRegisterEntryStatus.UNREGISTER)); + break; + } + for (int i = 0; i < chars.length; i++) { + if (chars[i] != '\0'){ + offset++; + } + if (chars[i] != '\n') { + data.append(chars[i]); + continue; + } + // HBase + System.out.println(data); + + 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; + } + data.delete(0, data.length()); + } + } + } 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); + } + + try { + Thread.sleep(SWITCH_FILE_WAIT_TIME); + } catch (InterruptedException e) { + logger.error("Failure sleep.", 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; + } + + 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 (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; + } +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThreadLauncher.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThreadLauncher.java new file mode 100644 index 000000000..5426f31ab --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThreadLauncher.java @@ -0,0 +1,13 @@ +package com.ai.cloud.skywalking.reciever.persistance; + +import com.ai.cloud.skywalking.reciever.conf.Config; + +public class PersistenceThreadLauncher { + + public static void doLaunch() { + new RegisterPersistenceThread().start(); + for (int i = 0; i < Config.Persistence.MAX_THREAD_NUMBER; i++) { + new PersistenceThread().start(); + } + } +} 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 new file mode 100644 index 000000000..d748cc32b --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/RegisterPersistenceThread.java @@ -0,0 +1,68 @@ +package com.ai.cloud.skywalking.reciever.persistance; + +import org.apache.commons.io.FileUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.Collection; + +import static com.ai.cloud.skywalking.reciever.conf.Config.RegisterPersistence.*; + +public class RegisterPersistenceThread extends Thread { + + private Logger logger = LogManager.getLogger(RegisterPersistenceThread.class); + + private BufferedWriter writer; + + public RegisterPersistenceThread() { + + } + + @Override + public void run() { + while (true) { + try { + Thread.sleep(OFFSET_WRITTEN_FILE_WAIT_CYCLE); + } 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 { + FileUtils.copyFile(file, bakFile); + } catch (IOException e) { + logger.error("Sleep failure", e); + } + Collection fileRegisterEntries = MemoryRegister.instance().getEntries(); + logger.info("file Register Entries size [{}]", fileRegisterEntries.size()); + BufferedWriter writer = null; + 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"); + } catch (IOException e) { + e.printStackTrace(); + } + } + try { + writer.write("EOF\n"); + writer.flush(); + writer.close(); + } catch (IOException e) { + e.printStackTrace(); + } + + } + } +} diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/ByteArrayUtil.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/ByteArrayUtil.java new file mode 100644 index 000000000..f727544bb --- /dev/null +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/util/ByteArrayUtil.java @@ -0,0 +1,16 @@ +package com.ai.cloud.skywalking.reciever.util; + +public class ByteArrayUtil { + + private ByteArrayUtil() { + } + + public static int byteArrayToInt(byte[] b, int offset) { + int value = 0; + for (int i = 0; i < 4; i++) { + int shift = (4 - 1 - i) * 8; + value += (b[i + offset] & 0x000000FF) << shift; + } + return value; + } +} diff --git a/skywalking-server/src/main/resources/config.properties b/skywalking-server/src/main/resources/config.properties new file mode 100644 index 000000000..c21ff8088 --- /dev/null +++ b/skywalking-server/src/main/resources/config.properties @@ -0,0 +1,35 @@ +#采集服务器的端口 +server.port=34000 + +#最大数据缓存线程数量 +buffer.max_thread_number=1 +#每个线程最大缓存数量 +buffer.per_thread_max_buffer_number=1024 +#无数据处理时轮询等待时间(单位:毫秒) +buffer.max_wait_time=5000 +#数据冲突时等待时间(单位:毫秒) +buffer.data_conflict_wait_time=10 +#数据缓存文件目录 +buffer.data_buffer_file_parent_directory=D:/test-data/data/buffer +#缓存数据文件最大长度(单位:byte) +buffer.data_file_max_length=10240 +#每次Flush的缓存数据的个数 +buffer.flush_number_of_cache=30 + +#最大持久化的线程数量 +persistence.max_thread_number=1 +#定位文件时,每次读取偏移量跳过大小 +persistence.offset_file_skip_length=2048 +#每次读取文件偏移量大小 +persistence.offset_file_read_buffer_size=2048 +#处理文件完成之后，等待时间(单位:毫秒) +persistence.switch_file_wait_time=5000 + +#偏移量注册文件的目录 +registerpersistence.register_file_parent_directory=D:/test-data/data/offset +#偏移量注册文件名 +registerpersistence.register_file_name=offset.txt +#偏移量注册备份文件名 +registerpersistence.register_bak_file_name=offset.txt.bak +#偏移量写入文件等待周期(单位:毫秒) +registerpersistence.offset_written_file_wait_cycle=5000 \ No newline at end of file diff --git a/skywalking-server/src/main/resources/log4j2.xml b/skywalking-server/src/main/resources/log4j2.xml new file mode 100644 index 000000000..025d5b2fb --- /dev/null +++ b/skywalking-server/src/main/resources/log4j2.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/skywalking-server/src/test/java/com/ai/cloud/skywalking/reciever/thread/PersistenceThreadTest.java b/skywalking-server/src/test/java/com/ai/cloud/skywalking/reciever/thread/PersistenceThreadTest.java new file mode 100644 index 000000000..e198e9b49 --- /dev/null +++ b/skywalking-server/src/test/java/com/ai/cloud/skywalking/reciever/thread/PersistenceThreadTest.java @@ -0,0 +1,36 @@ +package com.ai.cloud.skywalking.reciever.thread; + +import com.ai.cloud.skywalking.reciever.persistance.PersistenceThread; +import org.apache.commons.io.FileUtils; +import org.junit.Test; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; + +/** + * Created by astraea on 2015/11/5. + */ +public class PersistenceThreadTest { + + @Test + public void testDealDataFile() throws InterruptedException { + new PersistenceThread().start(); + Thread.sleep(500000L); + } + + @Test + public void test() { + System.out.println(Integer.valueOf(30 * 1024* 1024)); + } + + @Test + public void testFile() throws IOException { + File file = new File("D:\\test-data\\data\\buffer", "1446801421453-e10f3cc6279d48bebbd03fc3938ad665"); + FileWriter writer = new FileWriter(file,true); + writer.write("ssss"); + writer.flush(); + writer.close(); + System.out.println(FileUtils.deleteQuietly(file)); + } +} \ No newline at end of file diff --git a/skywalking-server/target/classes/config.properties b/skywalking-server/target/classes/config.properties new file mode 100644 index 000000000..c21ff8088 --- /dev/null +++ b/skywalking-server/target/classes/config.properties @@ -0,0 +1,35 @@ +#采集服务器的端口 +server.port=34000 + +#最大数据缓存线程数量 +buffer.max_thread_number=1 +#每个线程最大缓存数量 +buffer.per_thread_max_buffer_number=1024 +#无数据处理时轮询等待时间(单位:毫秒) +buffer.max_wait_time=5000 +#数据冲突时等待时间(单位:毫秒) +buffer.data_conflict_wait_time=10 +#数据缓存文件目录 +buffer.data_buffer_file_parent_directory=D:/test-data/data/buffer +#缓存数据文件最大长度(单位:byte) +buffer.data_file_max_length=10240 +#每次Flush的缓存数据的个数 +buffer.flush_number_of_cache=30 + +#最大持久化的线程数量 +persistence.max_thread_number=1 +#定位文件时,每次读取偏移量跳过大小 +persistence.offset_file_skip_length=2048 +#每次读取文件偏移量大小 +persistence.offset_file_read_buffer_size=2048 +#处理文件完成之后，等待时间(单位:毫秒) +persistence.switch_file_wait_time=5000 + +#偏移量注册文件的目录 +registerpersistence.register_file_parent_directory=D:/test-data/data/offset +#偏移量注册文件名 +registerpersistence.register_file_name=offset.txt +#偏移量注册备份文件名 +registerpersistence.register_bak_file_name=offset.txt.bak +#偏移量写入文件等待周期(单位:毫秒) +registerpersistence.offset_written_file_wait_cycle=5000 \ No newline at end of file diff --git a/skywalking-server/target/classes/log4j2.xml b/skywalking-server/target/classes/log4j2.xml new file mode 100644 index 000000000..025d5b2fb --- /dev/null +++ b/skywalking-server/target/classes/log4j2.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/skywalking-server/target/maven-archiver/pom.properties b/skywalking-server/target/maven-archiver/pom.properties new file mode 100644 index 000000000..58b9e313b --- /dev/null +++ b/skywalking-server/target/maven-archiver/pom.properties @@ -0,0 +1,5 @@ +#Generated by Maven +#Sat Nov 07 10:31:28 CST 2015 +version=1.0-SNAPSHOT +groupId=com.ai.cloud +artifactId=skywalking-server diff --git a/skywalking-server/target/maven-status/maven-compiler-plugin/compile/default-compile/createdFiles.lst b/skywalking-server/target/maven-status/maven-compiler-plugin/compile/default-compile/createdFiles.lst new file mode 100644 index 000000000..1ffb4ee25 --- /dev/null +++ b/skywalking-server/target/maven-status/maven-compiler-plugin/compile/default-compile/createdFiles.lst @@ -0,0 +1,18 @@ +com\ai\cloud\skywalking\reciever\util\ByteArrayUtil.class +com\ai\cloud\skywalking\reciever\conf\Config.class +com\ai\cloud\skywalking\reciever\persistance\RegisterPersistenceThread.class +com\ai\cloud\skywalking\reciever\buffer\DataBufferThreadContainer.class +com\ai\cloud\skywalking\reciever\persistance\PersistenceThread.class +com\ai\cloud\skywalking\reciever\persistance\FileRegisterEntry$FileRegisterEntryStatus.class +com\ai\cloud\skywalking\reciever\CollectionServer.class +com\ai\cloud\skywalking\reciever\buffer\DataBufferThread.class +com\ai\cloud\skywalking\reciever\conf\Config$Persistence.class +com\ai\cloud\skywalking\reciever\constants\ServerConstants.class +com\ai\cloud\skywalking\reciever\conf\ConfigDesc.class +com\ai\cloud\skywalking\reciever\conf\Config$RegisterPersistence.class +com\ai\cloud\skywalking\reciever\persistance\FileRegisterEntry.class +com\ai\cloud\skywalking\reciever\conf\ConfigInitializer.class +com\ai\cloud\skywalking\reciever\persistance\PersistenceThreadLauncher.class +com\ai\cloud\skywalking\reciever\persistance\MemoryRegister.class +com\ai\cloud\skywalking\reciever\conf\Config$Server.class +com\ai\cloud\skywalking\reciever\conf\Config$Buffer.class diff --git a/skywalking-server/target/maven-status/maven-compiler-plugin/compile/default-compile/inputFiles.lst b/skywalking-server/target/maven-status/maven-compiler-plugin/compile/default-compile/inputFiles.lst new file mode 100644 index 000000000..f28959253 --- /dev/null +++ b/skywalking-server/target/maven-status/maven-compiler-plugin/compile/default-compile/inputFiles.lst @@ -0,0 +1,12 @@ +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\persistance\PersistenceThread.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\conf\ConfigInitializer.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\util\ByteArrayUtil.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\constants\ServerConstants.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\CollectionServer.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\persistance\FileRegisterEntry.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\buffer\DataBufferThread.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\conf\Config.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\persistance\RegisterPersistenceThread.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\buffer\DataBufferThreadContainer.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\persistance\MemoryRegister.java +D:\iworkbench\PaaS-PerformanceMonitor-DEV\skywalking-server\src\main\java\com\ai\cloud\skywalking\reciever\persistance\PersistenceThreadLauncher.java