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

完成DataFileIndex功能

This commit is contained in:
ascrutae 2016-11-02 20:11:52 +08:00
parent 8f50a79e69
commit f2bdef7fa7
20 changed files with 416 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
skywalking-storage-center/skywalking-storage/pom.xml
View File

@ -61,37 +61,11 @@
<artifactId>HikariCP</artifactId>
<version>2.4.3</version>
</dependency>
<dependency>
<groupId>org.hsqldb</groupId>
<artifactId>hsqldb</artifactId>
<version>2.3.4</version>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-core</artifactId>
<version>1.6.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-module-junit4</artifactId>
<version>1.6.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.powermock</groupId>
<artifactId>powermock-api-mockito</artifactId>
<version>1.6.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>2.2.9</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
@ -184,4 +158,6 @@
</plugin>
</plugins>
</build>
</project>

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

@ -0,0 +1,13 @@
package com.a.eye.skywalking.storage.config;
/**
* Created by xin on 2016/11/2.
*/
public class Config {
public static class DataFileIndex {
public static String STORAGE_BASE_PATH = "/tmp/skywalking/index";
public static String DATA_FILE_INDEX_FILE_NAME = "data_file.index";
}
}

24
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/DataFileFinder.java Normal file
View File

@ -0,0 +1,24 @@
package com.a.eye.skywalking.storage.index;
/**
* Created by xin on 2016/11/2.
*/
public class DataFileFinder {
private IndexL1Cache l1Cache;
private IndexL2Cache l2Cache;
public DataFileFinder(IndexL1Cache l1Cache, IndexL2Cache l2Cache) {
this.l1Cache = l1Cache;
this.l2Cache = l2Cache;
}
public long find(long timestamp) {
Long index = l1Cache.find(timestamp);
if (index == null) {
index = l2Cache.find(timestamp);
}
return index;
}
}

22
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/DataFileIndexEngine.java
View File

@ -1,28 +1,24 @@
package com.a.eye.skywalking.storage.index;
import com.a.eye.skywalking.storage.index.cache.DataFileIndexCache;
/**
* 用于初始化DataFileIndex的数据，并且提供操作Index的所有类
*/
public class DataFileIndexEngine {
private static final DataFileIndexCache indexCache;
private static IndexL1Cache l1Cache;
private static IndexL2Cache l2Cache;
static {
indexCache = new DataFileIndexCache();
public static void start(){
l1Cache = new IndexL1Cache();
l2Cache = new IndexL2Cache();
newUpdator().init();
}
private DataFileIndexEngine() {
//DO Nothing
public static DataFileFinder newFinder() {
return new DataFileFinder(l1Cache, l2Cache);
}
public static DataFileIndexFinder newFinder() {
return indexCache;
}
public static DataFileIndexUpdator newUpdator() {
return indexCache;
return new DataFileIndexUpdator(l1Cache, l2Cache);
}
}

5
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/DataFileIndexFinder.java
View File

@ -1,5 +0,0 @@
package com.a.eye.skywalking.storage.index;
public interface DataFileIndexFinder {
DataIndexMetaData find(long timestamp);
}

88
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/DataFileIndexUpdator.java
View File

@ -1,5 +1,89 @@
package com.a.eye.skywalking.storage.index;
public interface DataFileIndexUpdator {
void update(long timestamp);
import com.a.eye.skywalking.storage.index.exception.DataFileIndexSaveFailedException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import static com.a.eye.skywalking.storage.config.Config.DataFileIndex.DATA_FILE_INDEX_FILE_NAME;
import static com.a.eye.skywalking.storage.config.Config.DataFileIndex.STORAGE_BASE_PATH;
public class DataFileIndexUpdator {
private static Logger logger = LogManager.getLogger(DataFileIndexUpdator.class);
private IndexL1Cache l1Cache;
private IndexL2Cache l2Cache;
public DataFileIndexUpdator(IndexL1Cache l1Cache, IndexL2Cache l2Cache) {
this.l1Cache = l1Cache;
this.l2Cache = l2Cache;
}
public void addRecord(long timestamp) {
logger.info("Updating index. timestamp:{}", timestamp);
try {
updateFile(timestamp);
updateCache(timestamp);
} catch (Exception e) {
logger.error("Failed to add index record", e);
}
}
private void updateCache(long timestamp) {
l1Cache.update(timestamp);
l2Cache.update(timestamp);
}
private void updateFile(long timestamp) throws DataFileIndexSaveFailedException {
BufferedWriter writer = null;
try {
writer = new BufferedWriter(new FileWriter(new File(STORAGE_BASE_PATH, DATA_FILE_INDEX_FILE_NAME)));
writer.write(String.valueOf(timestamp));
writer.newLine();
writer.close();
} catch (IOException e) {
throw new DataFileIndexSaveFailedException("Failed to save index[" + timestamp + "]", e);
} finally {
if (writer != null) {
try {
writer.close();
} catch (IOException e) {
logger.error("Failed to close index file", e);
}
}
}
}
void init() {
List<Long> indexData = new ArrayList<>();
BufferedReader indexFileReader = null;
try {
indexFileReader =
new BufferedReader(new FileReader(new File(STORAGE_BASE_PATH, DATA_FILE_INDEX_FILE_NAME)));
String indexDataStr = null;
while ((indexDataStr = indexFileReader.readLine()) != null) {
indexData.add(Long.parseLong(indexDataStr));
}
} catch (IOException e) {
logger.error("Failed to read index data.", e);
} finally {
if (indexFileReader != null) {
try {
indexFileReader.close();
} catch (IOException e) {
logger.error("Failed to close index file", e);
}
}
}
Collections.reverse(indexData);
l1Cache.initData(indexData);
l2Cache.initData(indexData);
}
}

7
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/DataIndexMetaData.java
View File

@ -1,7 +0,0 @@
package com.a.eye.skywalking.storage.index;
/**
* Created by xin on 2016/11/1.
*/
public class DataIndexMetaData {
}

56
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/IndexL1Cache.java Normal file
View File

@ -0,0 +1,56 @@
package com.a.eye.skywalking.storage.index;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* Created by xin on 2016/11/2.
*/
public class IndexL1Cache {
private static final int MAX_DATA_KEEP_SIZE = 30;
private static Logger logger = LogManager.getLogger(IndexL1Cache.class);
private TreeSet<Long> cacheData = new TreeSet<Long>();
private final ReadWriteLock updateLock = new ReentrantReadWriteLock();
void initData(List<Long> data) {
for (int i = 0; i < MAX_DATA_KEEP_SIZE; i++) {
this.cacheData.add(data.get(i));
}
}
public Long find(long timestamp) {
Lock lock = updateLock.readLock();
try {
lock.lock();
return cacheData.higher(timestamp);
} finally {
lock.unlock();
}
}
public void update(long timestamp) {
TreeSet<Long> newCacheData = new TreeSet<>(cacheData);
newCacheData.add(timestamp);
if (newCacheData.size() >= MAX_DATA_KEEP_SIZE + 1) {
long removedData = newCacheData.pollFirst();
logger.info("Add cache data : {}, removed cache Data:{}", timestamp, removedData);
}
Lock lock = updateLock.writeLock();
try {
lock.lock();
cacheData = newCacheData;
} finally {
lock.unlock();
}
}
}

40
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/IndexL2Cache.java Normal file
View File

@ -0,0 +1,40 @@
package com.a.eye.skywalking.storage.index;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class IndexL2Cache {
private TreeSet<Long> cacheData = new TreeSet<Long>();
private ReadWriteLock updateLock = new ReentrantReadWriteLock();
void initData(List<Long> data) {
this.cacheData.addAll(cacheData);
}
public long find(long timestamp) {
Lock lock = updateLock.readLock();
try {
lock.lock();
return this.cacheData.higher(timestamp);
} finally {
lock.unlock();
}
}
public void update(long timestamp) {
TreeSet<Long> newCacheData = new TreeSet<>(cacheData);
newCacheData.add(timestamp);
Lock lock = updateLock.writeLock();
try {
lock.lock();
cacheData.add(timestamp);
} finally {
lock.unlock();
}
}
}

26
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/cache/DataFileIndexCache.java vendored
View File

@ -1,26 +0,0 @@
package com.a.eye.skywalking.storage.index.cache;
import com.a.eye.skywalking.storage.index.DataFileIndexFinder;
import com.a.eye.skywalking.storage.index.DataFileIndexUpdator;
import com.a.eye.skywalking.storage.index.DataIndexMetaData;
public class DataFileIndexCache implements DataFileIndexFinder, DataFileIndexUpdator {
private IndexL1Cache l1Cache;
private IndexL2Cache l2Cache;
public DataFileIndexCache() {
//load the index data
}
@Override
public void update(long timestamp) {
// TODO: 2016/11/1 通知Cache更新，以及Index的文件更新
}
@Override
public DataIndexMetaData find(long timestamp) {
// TODO: 2016/11/1 查找一级缓存，二级缓存
return null;
}
}

9
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/cache/DataFileIndexReader.java vendored
View File

@ -1,9 +0,0 @@
package com.a.eye.skywalking.storage.index.cache;
import java.util.SortedSet;
public class DataFileIndexReader {
public SortedSet<Long> read() {
return null;
}
}

8
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/cache/DataFileIndexWriter.java vendored
View File

@ -1,8 +0,0 @@
package com.a.eye.skywalking.storage.index.cache;
public class DataFileIndexWriter {
public void write(long timestamp) {
}
}

7
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/cache/IndexL1Cache.java vendored
View File

@ -1,7 +0,0 @@
package com.a.eye.skywalking.storage.index.cache;
/**
* Created by xin on 2016/11/1.
*/
class IndexL1Cache {
}

7
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/cache/IndexL2Cache.java vendored
View File

@ -1,7 +0,0 @@
package com.a.eye.skywalking.storage.index.cache;
/**
* Created by xin on 2016/11/1.
*/
class IndexL2Cache {
}

10
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/index/exception/DataFileIndexSaveFailedException.java Normal file
View File

@ -0,0 +1,10 @@
package com.a.eye.skywalking.storage.index.exception;
/**
* Created by xin on 2016/11/2.
*/
public class DataFileIndexSaveFailedException extends Exception {
public DataFileIndexSaveFailedException(String message, Exception e) {
super(message, e);
}
}

92
skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/search/HyperSqlSearchSpeedReporter.java Normal file
View File

@ -0,0 +1,92 @@
package com.a.eye.skywalking.search;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
/**
* Created by xin on 2016/11/1.
*/
public class HyperSqlSearchSpeedReporter {
private static final long BASE_TIME_STAMP = 1477983548L;
private static boolean useSingConnection = true;
private static HikariDataSource hikariDataSource;
private static Connection connection;
private static String CREATE_TABLE_SQL =
"CREATE TABLE data_index\n" + "(\n" + " id INT IDENTITY PRIMARY KEY NOT NULL,\n"
+ " startTime BIGINT NOT NULL\n" + ");\n";
private static String CREATE_INDEX_SQL =
"CREATE UNIQUE INDEX \"table_name_startTime_uindex\" ON data_index (startTime);";
private static String INSERT_DATA_SQL = "INSERT INTO data_index(startTime) VALUES(?);";
private static String QUERY_DATA_SQL =
"SELECT startTime FROM data_index WHERE startTime > ? ORDER BY startTime" + " ASC LIMIT 1";
public static void initData() throws SQLException {
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:hsqldb:mem:test-speed");
config.setUsername("root");
config.setPassword("root");
hikariDataSource = new HikariDataSource(config);
connection = hikariDataSource.getConnection();
PreparedStatement ps = connection.prepareStatement(CREATE_TABLE_SQL);
ps.execute();
ps = connection.prepareStatement(CREATE_INDEX_SQL);
ps.execute();
ps = connection.prepareStatement(INSERT_DATA_SQL);
for (int i = 0; i < 3000; i++) {
ps.setLong(1, BASE_TIME_STAMP + i * 1000 * 60 * 60);
//System.out.print(BASE_TIME_STAMP + i * 1000 * 60 * 60);
//System.out.print(",");
ps.execute();
}
//System.out.println();
ps.close();
}
public static long find(long element) throws SQLException {
Connection connection = null;
if (!useSingConnection) {
connection = hikariDataSource.getConnection();
}else{
connection = HyperSqlSearchSpeedReporter.connection;
}
PreparedStatement preparedStatement = connection.prepareStatement(QUERY_DATA_SQL);
preparedStatement.setLong(1, element);
ResultSet resultSet = preparedStatement.executeQuery();
resultSet.next();
long result = resultSet.getLong("startTime");
preparedStatement.close();
if (!useSingConnection){
connection.close();
}
return result;
}
public static void main(String[] args) throws SQLException {
initData();
long startTime = System.nanoTime();
for (long i = 0; i < 100000000L; i++) {
find(1478323448L);
}
long totalTime = System.nanoTime() - startTime;
System.out.println("total time : " + totalTime + " " + (totalTime * 1.0 / 100000000L));
}
}

37
skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/search/SearchSpeedReporter.java Normal file
View File

@ -0,0 +1,37 @@
package com.a.eye.skywalking.search;
import java.util.Arrays;
import java.util.TreeSet;
public class SearchSpeedReporter {
private static final long BASE_TIME_STAMP = 1477983548L;
private static Long[] testedData = new Long[3000];
private static TreeSet<Long> tree = new TreeSet<Long>();
public static void initData() {
for (int i = 0; i < 3000; i++) {
testedData[i] = new Long(BASE_TIME_STAMP + i * 1000 * 60 * 60L);
}
tree.addAll(Arrays.<Long>asList(testedData));
}
public static long find(long toElement) {
return tree.higher(toElement);
}
public static void main(String[] args) {
initData();
long startTime = System.nanoTime();
for (long i = 0; i < 100000000L; i++) {
find(1478323448L);
}
long totalTime = System.nanoTime() - startTime;
System.out.println("total time : " + totalTime + " " + (totalTime * 1.0 / 100000000L));
}
}

40
skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/search/TreeSetTest.java Normal file
View File

@ -0,0 +1,40 @@
package com.a.eye.skywalking.search;
import org.junit.Before;
import org.junit.Test;
import java.util.TreeSet;
import static org.junit.Assert.assertEquals;
/**
* Created by xin on 2016/11/1.
*/
public class TreeSetTest {
private TreeSet<Long> treeSet = new TreeSet<Long>();
@Before
public void setup(){
treeSet.add(9L);
treeSet.add(3L);
treeSet.add(13L);
treeSet.add(15L);
treeSet.add(1L);
treeSet.add(11L);
treeSet.add(5L);
treeSet.add(7L);
}
@Test
public void testGetElement(){
assertEquals(new Long(5), treeSet.higher(4L));
}
@Test
public void testRemoveElement(){
assertEquals(new Long(1), treeSet.first());
treeSet.pollFirst();
assertEquals(new Long(3), treeSet.first());
}
}

7
skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/storage/index/DataIndexFileOperator.java Normal file
View File

@ -0,0 +1,7 @@
package com.a.eye.skywalking.storage.index;
/**
* Created by xin on 2016/10/31.
*/
public class DataIndexFileOperator {
}

56
skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/storage/index/HyperSQLConnectorPoolTest.java
View File

@ -1,56 +0,0 @@
package com.a.eye.skywalking.storage.index;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.runners.MockitoJUnitRunner;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/**
* Created by xin on 2016/10/31.
*/
@RunWith(MockitoJUnitRunner.class)
@PrepareForTest({HyperSQLConnector.class, HyperSQLConnectorPool.class})
public class HyperSQLConnectorPoolTest {
@Before
public void setUp() throws Exception {
}
@After
public void tearDown() throws Exception {
}
@Test
public void getOrCreateConnector() throws Exception {
}
@Test
public void createConnector() throws Exception {
for (int i = 0; i < 100; i++) {
long index = System.currentTimeMillis();
HyperSQLConnector connector = HyperSQLConnectorPool.createConnector(index);
assertNotNull(connector);
if (i < 30) {
assertEquals(i + 1, HyperSQLConnectorPool.getCurrentConnectorsSize());
} else {
assertEquals(30, HyperSQLConnectorPool.getCurrentConnectorsSize());
}
assertEquals(HyperSQLConnectorPool.getIndex()[0], index);
}
}
}