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

完成写入数据和读取数据功能

This commit is contained in:
ascrutae 2016-11-13 19:43:15 +08:00
parent e6f5e78fd9
commit fc6ecf1162
10 changed files with 110 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/Main.java
View File

@ -9,6 +9,7 @@ import com.a.eye.skywalking.registry.RegistryCenterFactory;
import com.a.eye.skywalking.registry.api.CenterType;
import com.a.eye.skywalking.registry.api.RegistryCenter;
import com.a.eye.skywalking.registry.impl.zookeeper.ZookeeperConfig;
import com.a.eye.skywalking.storage.block.index.BlockIndexEngine;
import com.a.eye.skywalking.storage.config.Config;
import com.a.eye.skywalking.storage.config.ConfigInitializer;
import com.a.eye.skywalking.storage.data.IndexDataCapacityMonitor;
@ -37,21 +38,24 @@ public class Main {
public static void main(String[] args) {
try {
initializeParam();
new Thread(new IndexDataCapacityMonitor()).start();
BlockIndexEngine.start();
IndexDataCapacityMonitor.start();
transferService =
TransferServiceBuilder.newBuilder(Config.Server.PORT).startSpanStorageService(new StorageNotifier())
.startTraceSearchService(new SearchNotifier()).build();
transferService.start();
logger.info("transfer service started successfully!");
logger.info("transfer service started successfully.");
registryNode();
logger.info("storage service started successfully!");
logger.info("storage service started successfully.");
Thread.currentThread().join();
} catch (Throwable e) {
e.printStackTrace();
logger.error("Failed to start service.", e);
} finally {
transferService.stop();
IndexDataCapacityMonitor.stop();
}
}

25
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockIndexUpdator.java
View File

@ -42,7 +42,8 @@ public class BlockIndexUpdator {
private void updateFile(long timestamp) throws BlockIndexPersistenceFailedException {
BufferedWriter writer = null;
try {
writer = new BufferedWriter(new FileWriter(new File(STORAGE_BASE_PATH, DATA_FILE_INDEX_FILE_NAME)));
File blockIndexFile = getOrCreateBlockIndexFile();
writer = new BufferedWriter(new FileWriter(blockIndexFile));
writer.write(String.valueOf(timestamp));
writer.newLine();
writer.close();
@ -64,7 +65,8 @@ public class BlockIndexUpdator {
List<Long> indexData = new ArrayList<>();
BufferedReader indexFileReader = null;
try {
indexFileReader = new BufferedReader(new FileReader(new File(STORAGE_BASE_PATH, DATA_FILE_INDEX_FILE_NAME)));
File blockIndexFile = getOrCreateBlockIndexFile();
indexFileReader = new BufferedReader(new FileReader(blockIndexFile));
String indexDataStr = null;
while ((indexDataStr = indexFileReader.readLine()) != null) {
indexData.add(Long.parseLong(indexDataStr));
@ -81,8 +83,27 @@ public class BlockIndexUpdator {
}
}
if (indexData.size() == 0) {
//如果此前没有记录，则取之前五分钟到目前的数据
addRecord(System.currentTimeMillis() - 5 * 60 * 1000);
return;
}
Collections.reverse(indexData);
l1Cache.init(indexData);
l2Cache.init(indexData);
}
public File getOrCreateBlockIndexFile() throws IOException {
File blockIndexFile = new File(STORAGE_BASE_PATH, DATA_FILE_INDEX_FILE_NAME);
if (!blockIndexFile.getParentFile().exists()) {
blockIndexFile.getParentFile().mkdirs();
}
if (!blockIndexFile.exists()) {
blockIndexFile.createNewFile();
}
return blockIndexFile;
}
}

3
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/L1Cache.java
View File

@ -22,7 +22,8 @@ public class L1Cache {
private final ReadWriteLock updateLock = new ReentrantReadWriteLock();
void init(List<Long> data) {
for (int i = 0; i < MAX_DATA_KEEP_SIZE; i++) {
int size = data.size() > MAX_DATA_KEEP_SIZE ? MAX_DATA_KEEP_SIZE : data.size();
for (int i = 0; i < size; i++) {
this.cacheData.add(data.get(i));
}
}

2
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/L2Cache.java
View File

@ -15,7 +15,7 @@ public class L2Cache {
private ReadWriteLock updateLock = new ReentrantReadWriteLock();
void init(List<Long> data) {
this.cacheData.addAll(cacheData);
this.cacheData.addAll(data);
}
public Long find(long timestamp) {

13
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java
View File

@ -7,25 +7,30 @@ public class Constants {
public static class SQL {
public static final String CREATE_TABLE = "CREATE TABLE " + TABLE_NAME + "\n" + "(\n"
+ " id INT PRIMARY KEY NOT NULL IDENTITY,\n"
+ " trace_id VARCHAR(32) NOT NULL,\n"
+ " trace_id VARCHAR(64) NOT NULL,\n"
+ " levelId VARCHAR(1024) NOT NULL,\n"
+ " span_type INT NOT NULL, \n"
+ " file_name VARCHAR(10) NOT NULL,\n"
+ " file_name VARCHAR(32) NOT NULL,\n"
+ " offset BIGINT NOT NULL,\n"
+ " length INT NOT NULL\n" + ");\n";
public static final String CREATE_INDEX = "CREATE INDEX \"index_data_trace_id_index\" ON " + TABLE_NAME + " (trace_id);";
public static final String INSERT_INDEX = "INSERT INTO " +TABLE_NAME + "(trace_id,levelId,span_type"
+ "file_name,offset,length) VALUES(?,?,?,?,?,?)";
+ ",file_name,offset,length) VALUES(?,?,?,?,?,?)";
public static final String QUERY_TABLES = "SELECT count(1) AS TABLE_COUNT FROM INFORMATION_SCHEMA.TABLES "
+ "WHERE TABLE_NAME= '" + TABLE_NAME + "';";
+ "WHERE TABLE_NAME= '" + TABLE_NAME.toUpperCase() + "';";
public static final String QUERY_INDEX_SIZE = "SELECT count(1) AS INDEX_SIZE FROM " + TABLE_NAME;
public static final String QUERY_TRACE_ID = "SELECT span_type, file_name, offset, length "
+ " FROM "+ TABLE_NAME+ " WHERE trace_id = ?";
public static final String DEFAULT_USER = "root";
public static final String DEFAULT_PASSWORD = "root";
}
}

37
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java
View File

@ -6,6 +6,9 @@ import com.a.eye.skywalking.storage.block.index.BlockIndexEngine;
import com.a.eye.skywalking.storage.data.index.IndexDBConnector;
import java.sql.SQLException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static com.a.eye.skywalking.storage.config.Config.DataIndex.MAX_CAPACITY_PER_INDEX;
@ -13,7 +16,7 @@ import static com.a.eye.skywalking.storage.config.Config.DataIndex.MAX_CAPACITY_
/**
* Created by xin on 2016/11/6.
*/
public class IndexDataCapacityMonitor implements Runnable {
public class IndexDataCapacityMonitor {
private static ILog logger = LogManager.getLogger(IndexDataCapacityMonitor.class);
private static Detector detector;
@ -24,19 +27,30 @@ public class IndexDataCapacityMonitor implements Runnable {
}
}
private class Detector {
private static class Detector extends TimerTask {
private AtomicLong currentSize;
private long timestamp;
private Timer timer = new Timer();
public Detector(long timestamp) {
this.timestamp = timestamp;
currentSize = new AtomicLong();
startTimer();
}
public Detector(long timestamp, long currentSize) {
this.currentSize = new AtomicLong(currentSize);
this.timestamp = timestamp;
startTimer();
}
public void startTimer() {
timer.scheduleAtFixedRate(this, 0, TimeUnit.SECONDS.toMillis(30));
}
public void stopTimer() {
timer.cancel();
}
public boolean isDetectFor(long timestamp) {
@ -44,26 +58,31 @@ public class IndexDataCapacityMonitor implements Runnable {
}
public void add(int updateRecordSize) {
if (currentSize.addAndGet(updateRecordSize) > MAX_CAPACITY_PER_INDEX * 0.8) {
currentSize.addAndGet(updateRecordSize);
}
@Override
public void run() {
if (currentSize.get() > MAX_CAPACITY_PER_INDEX * 0.8) {
notificationAddNewBlockIndexAndCreateNewIndexDB();
stopTimer();
}
}
}
private void notificationAddNewBlockIndexAndCreateNewIndexDB() {
private static void notificationAddNewBlockIndexAndCreateNewIndexDB() {
long timestamp = System.currentTimeMillis() + 5 * 60 * 1000;
BlockIndexEngine.newUpdator().addRecord(timestamp);
createNewIndexDB(timestamp);
detector = new Detector(timestamp);
}
private void createNewIndexDB(long timestamp) {
private static void createNewIndexDB(long timestamp) {
IndexDBConnector connector = new IndexDBConnector(timestamp);
connector.close();
}
@Override
public void run() {
public static void start() {
long timestamp = BlockIndexEngine.newFinder().findLastBlockIndex();
IndexDBConnector dbConnector = null;
@ -84,4 +103,8 @@ public class IndexDataCapacityMonitor implements Runnable {
}
}
public static void stop() {
detector.stopTimer();
}
}

5
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataFinder.java
View File

@ -13,6 +13,9 @@ import com.zaxxer.hikari.HikariDataSource;
import java.sql.SQLException;
import java.util.*;
import static com.a.eye.skywalking.storage.config.Constants.SQL.DEFAULT_PASSWORD;
import static com.a.eye.skywalking.storage.config.Constants.SQL.DEFAULT_USER;
public class SpanDataFinder {
private static ILog logger = LogManager.getLogger(SpanDataFinder.class);
private static IndexDataSourceCache datasourceCache = new IndexDataSourceCache(Config.SpanFinder.MAX_CACHE_SIZE);
@ -71,6 +74,8 @@ public class SpanDataFinder {
config.setJdbcUrl(new ConnectURLGenerator(Config.DataIndex.BASE_PATH, Config.DataIndex.STORAGE_INDEX_FILE_NAME)
.generate(blockIndex));
config.setDriverClassName("org.hsqldb.jdbc.JDBCDriver");
config.setUsername(DEFAULT_USER);
config.setPassword(DEFAULT_PASSWORD);
config.setMaximumPoolSize(20);
config.setMinimumIdle(5);
return config;

33
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java
View File

@ -1,11 +1,13 @@
package com.a.eye.skywalking.storage.data.file;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.storage.config.Config;
import com.a.eye.skywalking.storage.data.spandata.SpanData;
import com.a.eye.skywalking.storage.data.exception.DataFileOperatorCreateFailedException;
import com.a.eye.skywalking.storage.data.exception.SpanDataPersistenceFailedException;
import com.a.eye.skywalking.storage.data.exception.SpanDataReadFailedException;
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
import com.a.eye.skywalking.storage.data.spandata.SpanData;
import java.io.File;
import java.io.FileInputStream;
@ -17,25 +19,48 @@ import java.io.IOException;
*/
public class DataFile {
private static ILog logger = LogManager.getLogger(DataFile.class);
private String fileName;
private long currentOffset;
private DataFileOperator operator;
static {
File dataFileDir = new File(Config.DataFile.BASE_PATH);
if (!dataFileDir.exists()) {
dataFileDir.mkdirs();
}
}
public DataFile() {
this.fileName = System.currentTimeMillis() + "";
this.currentOffset = 0;
operator = new DataFileOperator();
createFile();
}
public DataFile(String fileName) {
this.fileName = fileName;
operator = new DataFileOperator();
createFile();
}
public DataFile(File file) {
this.fileName = file.getName();
this.currentOffset = file.length();
operator = new DataFileOperator();
createFile();
}
private void createFile() {
File dataFile = new File(Config.DataFile.BASE_PATH, fileName);
if (!dataFile.exists()) {
try {
dataFile.createNewFile();
} catch (IOException e) {
logger.error("Failed to create data file.", e);
throw new DataFileOperatorCreateFailedException("Failed to create data file", e);
}
}
}
public boolean overLimitLength() {
@ -46,7 +71,7 @@ public class DataFile {
byte[] bytes = data.toByteArray();
try {
operator.getWriter().write(bytes);
IndexMetaInfo metaInfo = new IndexMetaInfo(data,fileName, currentOffset, bytes.length);
IndexMetaInfo metaInfo = new IndexMetaInfo(data, fileName, currentOffset, bytes.length);
currentOffset += bytes.length;
return metaInfo;
} catch (IOException e) {
@ -82,7 +107,7 @@ public class DataFile {
if (writer == null) {
try {
writer = new FileOutputStream(new File(fileName));
writer = new FileOutputStream(new File(Config.DataFile.BASE_PATH, fileName), true);
} catch (IOException e) {
throw new DataFileOperatorCreateFailedException("Failed to create datafile output stream", e);
}
@ -94,7 +119,7 @@ public class DataFile {
public FileInputStream getReader() {
if (reader == null) {
try {
reader = new FileInputStream(new File(fileName));
reader = new FileInputStream(new File(Config.DataFile.BASE_PATH, fileName));
} catch (IOException e) {
throw new DataFileOperatorCreateFailedException("Failed to create datafile input stream", e);
}

6
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java
View File

@ -48,7 +48,7 @@ public class IndexDBConnector {
private void createTableAndIndexIfNecessary() {
try {
if (validateTableIsExists()) {
if (!tableExists()) {
createTable();
createIndex();
}
@ -60,14 +60,14 @@ public class IndexDBConnector {
private void createConnection() {
try {
connection = DriverManager.getConnection(generator.generate(timestamp));
connection = DriverManager.getConnection(generator.generate(timestamp), DEFAULT_USER, DEFAULT_PASSWORD);
connection.setAutoCommit(true);
} catch (SQLException e) {
throw new ConnectorInitializeFailedException("Failed to create connection.", e);
}
}
private boolean validateTableIsExists() throws SQLException {
private boolean tableExists() throws SQLException {
PreparedStatement ps = connection.prepareStatement(QUERY_TABLES);
ResultSet rs = ps.executeQuery();
rs.next();

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

@ -29,7 +29,7 @@ registrycenter.auth_info=
#
registrycenter.auth_schema=
#
registrycenter.connect_url=
registrycenter.connect_url= 127.0.0.1:2181
#
registrycenter.registry_path_prefix=
#