完成DataFile功能
This commit is contained in:
ascrutae
parent
24e9af5152
commit
32777d72c7
|
|
@ -18,6 +18,11 @@ public class BlockFinder {
|
|||
if (index == null) {
|
||||
index = l2Cache.find(timestamp);
|
||||
}
|
||||
|
||||
if (index == null) {
|
||||
index = System.currentTimeMillis();
|
||||
}
|
||||
|
||||
return index;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -9,8 +9,8 @@ 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;
|
||||
import static com.a.eye.skywalking.storage.config.Config.BlockIndex.DATA_FILE_INDEX_FILE_NAME;
|
||||
import static com.a.eye.skywalking.storage.config.Config.BlockIndex.STORAGE_BASE_PATH;
|
||||
|
||||
public class BlockIndexUpdator {
|
||||
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ public class L2Cache {
|
|||
this.cacheData.addAll(cacheData);
|
||||
}
|
||||
|
||||
public long find(long timestamp) {
|
||||
public Long find(long timestamp) {
|
||||
Lock lock = updateLock.readLock();
|
||||
try {
|
||||
lock.lock();
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ package com.a.eye.skywalking.storage.config;
|
|||
* Created by xin on 2016/11/2.
|
||||
*/
|
||||
public class Config {
|
||||
public static class DataFileIndex {
|
||||
public static class BlockIndex {
|
||||
|
||||
public static String STORAGE_BASE_PATH = "/tmp/skywalking/index";
|
||||
|
||||
|
|
@ -14,5 +14,12 @@ public class Config {
|
|||
|
||||
public static class DataFile {
|
||||
public static String BASE_PATH = "";
|
||||
|
||||
public static long MAX_LENGTH = 3 * 1024 * 1024 * 1024;
|
||||
}
|
||||
|
||||
|
||||
public static class DataIndex {
|
||||
public static String TABLE_NAME = "data_index";
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,28 @@
|
|||
package com.a.eye.skywalking.storage.data;
|
||||
|
||||
import com.a.eye.skywalking.storage.block.index.BlockFinder;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
public class IndexMetaInfoCategory {
|
||||
|
||||
public static Map<Long, List<IndexMetaInfo>> categorizeByDataIndexTime(List<IndexMetaInfo> indexMetaInfo,
|
||||
BlockFinder finder) {
|
||||
Map<Long, List<IndexMetaInfo>> categorizeMetaInfo = new HashMap<Long, List<IndexMetaInfo>>();
|
||||
|
||||
for (IndexMetaInfo info : indexMetaInfo) {
|
||||
long timestamp = finder.find(info.getStartTime());
|
||||
|
||||
List<IndexMetaInfo> metaInfos = categorizeMetaInfo.get(timestamp);
|
||||
if (metaInfos == null) {
|
||||
metaInfos.add(info);
|
||||
categorizeMetaInfo.put(timestamp, metaInfos);
|
||||
}
|
||||
}
|
||||
|
||||
return categorizeMetaInfo;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,10 +1,8 @@
|
|||
package com.a.eye.skywalking.storage.data;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public interface SpanData {
|
||||
long getTimestamp();
|
||||
|
||||
byte[] convertToByte();
|
||||
long getStartTime();
|
||||
|
||||
byte[] toByteArray();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +0,0 @@
|
|||
package com.a.eye.skywalking.storage.data;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class SpanDataBuilder {
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package com.a.eye.skywalking.storage.data;
|
||||
|
||||
import com.a.eye.datacarrier.consumer.IConsumer;
|
||||
import com.a.eye.skywalking.storage.block.index.BlockIndexEngine;
|
||||
import com.a.eye.skywalking.storage.data.file.DataFileWriter;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexOperator;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexOperatorCache;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
public class SpanDataConsumer implements IConsumer<SpanData> {
|
||||
|
||||
private IndexOperatorCache cache;
|
||||
private DataFileWriter fileWriter;
|
||||
|
||||
@Override
|
||||
public void consume(List<SpanData> data) {
|
||||
List<IndexMetaInfo> indexMetaInfo = fileWriter.write(data);
|
||||
|
||||
Map<Long, List<IndexMetaInfo>> categorizedMetaInfo =
|
||||
IndexMetaInfoCategory.categorizeByDataIndexTime(indexMetaInfo, BlockIndexEngine.newFinder());
|
||||
|
||||
for (Map.Entry<Long, List<IndexMetaInfo>> indexEntry : categorizedMetaInfo.entrySet()) {
|
||||
IndexOperator indexOperator = cache.get(indexEntry.getKey());
|
||||
indexOperator.update(indexEntry.getValue());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(List<SpanData> list, Throwable throwable) {
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -1,28 +0,0 @@
|
|||
package com.a.eye.skywalking.storage.data;
|
||||
|
||||
import com.a.eye.datacarrier.consumer.IConsumer;
|
||||
import com.a.eye.skywalking.storage.data.file.DataFileWriter;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexOperator;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexOperatorFactory;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public class SpanDataWriterConsumer implements IConsumer<SpanData> {
|
||||
|
||||
private DataFileWriter writer;
|
||||
|
||||
@Override
|
||||
public void consume(List<SpanData> list) {
|
||||
for (SpanData data : list) {
|
||||
IndexOperator operator = IndexOperatorFactory.get(data.getTimestamp());
|
||||
IndexMetaInfo metaInfo = writer.write(data.convertToByte());
|
||||
operator.update(metaInfo);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(List<SpanData> list, Throwable throwable) {
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -1,10 +0,0 @@
|
|||
package com.a.eye.skywalking.storage.data.exception;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class DataFileNotFoundException extends Exception {
|
||||
public DataFileNotFoundException(String message, Exception e) {
|
||||
super(message, e);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,10 +0,0 @@
|
|||
package com.a.eye.skywalking.storage.data.exception;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class FileReaderCreateFailedException extends Throwable {
|
||||
public FileReaderCreateFailedException(String message, Exception e) {
|
||||
super(message, e);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,27 +1,10 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
/**
|
||||
* 数据文件
|
||||
*/
|
||||
public class DataFile {
|
||||
|
||||
private static final long MAX_LENGTH = 3 * 1024 * 1024 * 1024;
|
||||
|
||||
private String name;
|
||||
private long currentLength;
|
||||
|
||||
public boolean overLimitLength() {
|
||||
return currentLength > MAX_LENGTH;
|
||||
}
|
||||
|
||||
|
||||
public long writeAndFlush(byte[] data) {
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void close() {
|
||||
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,17 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class DataFileLoader {
|
||||
|
||||
public DataFileLoader(String basePath) {
|
||||
|
||||
}
|
||||
|
||||
public List<DataFile> load() {
|
||||
return new ArrayList<DataFile>();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -1,20 +1,10 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import com.a.eye.skywalking.storage.data.exception.DataFileNotFoundException;
|
||||
import com.a.eye.skywalking.storage.data.exception.FileReaderCreateFailedException;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
|
||||
|
||||
public class DataFileOperatorFactory {
|
||||
|
||||
public static DataFileReader newReader(IndexMetaInfo indexMetaInfo) throws FileReaderCreateFailedException {
|
||||
try {
|
||||
return new DataFileReader(indexMetaInfo);
|
||||
} catch (DataFileNotFoundException e) {
|
||||
throw new FileReaderCreateFailedException("Cannot create DataFileReader.", e);
|
||||
}
|
||||
}
|
||||
|
||||
public static DataFileWriter newWriter() {
|
||||
return new DataFileWriter();
|
||||
public static DataFileReader getDataFileReader(IndexMetaInfo info) {
|
||||
return new DataFileReader(info);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,45 +1,17 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import com.a.eye.skywalking.storage.config.Config;
|
||||
import com.a.eye.skywalking.storage.data.exception.DataFileNotFoundException;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileInputStream;
|
||||
import java.io.FileNotFoundException;
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class DataFileReader {
|
||||
|
||||
private static Logger logger = LogManager.getLogger(DataFileReader.class);
|
||||
public DataFileReader(IndexMetaInfo info) {
|
||||
|
||||
private long offset;
|
||||
private int length;
|
||||
private String fileName;
|
||||
private FileInputStream reader;
|
||||
|
||||
public DataFileReader(IndexMetaInfo indexMetaInfo) throws DataFileNotFoundException {
|
||||
try {
|
||||
reader = new FileInputStream(new File(Config.DataFile.BASE_PATH, indexMetaInfo.getFileName()));
|
||||
this.offset = indexMetaInfo.getOffset();
|
||||
this.length = indexMetaInfo.getLength();
|
||||
this.fileName = indexMetaInfo.getFileName();
|
||||
} catch (FileNotFoundException e) {
|
||||
throw new DataFileNotFoundException(indexMetaInfo.getFileName() + " not found.", e);
|
||||
}
|
||||
}
|
||||
|
||||
public byte[] read() {
|
||||
try {
|
||||
reader.getChannel().position(this.offset);
|
||||
byte[] dataByte = new byte[length];
|
||||
reader.read(dataByte, 0, length);
|
||||
return dataByte;
|
||||
} catch (IOException e) {
|
||||
logger.error("Failed to read file:{} position:{} length:{}", fileName, offset, length);
|
||||
return null;
|
||||
}
|
||||
return new byte[0];
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,27 +1,24 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import com.a.eye.skywalking.storage.data.SpanData;
|
||||
import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public class DataFileWriter {
|
||||
|
||||
private DataFile dataFile;
|
||||
|
||||
public DataFileWriter() {
|
||||
|
||||
dataFile = DataFilesManager.createNewDataFile();
|
||||
}
|
||||
|
||||
public IndexMetaInfo write(byte[] data) {
|
||||
public List<IndexMetaInfo> write(List<SpanData> spanData) {
|
||||
|
||||
if (dataFile.overLimitLength()) {
|
||||
convertDataFile();
|
||||
dataFile = DataFilesManager.createNewDataFile();
|
||||
}
|
||||
|
||||
long offset = dataFile.writeAndFlush(data);
|
||||
return new IndexMetaInfo(dataFile.getName(), offset, data.length);
|
||||
return null;
|
||||
}
|
||||
|
||||
private void convertDataFile() {
|
||||
dataFile.close();
|
||||
dataFile = new DataFile();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,29 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import com.a.eye.skywalking.storage.config.Config;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ConcurrentLinkedDeque;
|
||||
|
||||
/**
|
||||
* 管理数据文件，目前主要是用来加载所有未满的数据文件，以及创建数据文件
|
||||
*/
|
||||
public class DataFilesManager {
|
||||
|
||||
private static ConcurrentLinkedDeque<DataFile> unFinishedDataFiles = new ConcurrentLinkedDeque<>();
|
||||
|
||||
public static void init() {
|
||||
List<DataFile> allDataFile = new DataFileLoader(Config.DataFile.BASE_PATH).load();
|
||||
unFinishedDataFiles = new ConcurrentLinkedDeque<>(new UnFinishedDataFilePicker(allDataFile).pickUp());
|
||||
}
|
||||
|
||||
public static DataFile createNewDataFile() {
|
||||
DataFile dataFile = unFinishedDataFiles.poll();
|
||||
if (dataFile == null) {
|
||||
dataFile = new DataFile();
|
||||
}
|
||||
|
||||
return dataFile;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,20 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 用于选择所有没有达到最大容量的数据文件
|
||||
*/
|
||||
public class UnFinishedDataFilePicker {
|
||||
|
||||
private List<DataFile> dataFiles;
|
||||
|
||||
public UnFinishedDataFilePicker(List<DataFile> dataFiles) {
|
||||
|
||||
}
|
||||
|
||||
public List<DataFile> pickUp() {
|
||||
return new ArrayList<DataFile>();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,9 +1,7 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/3.
|
||||
*/
|
||||
public class IndexMetaInfo {
|
||||
private String traceId;
|
||||
|
||||
private String fileName;
|
||||
|
||||
|
|
@ -11,12 +9,7 @@ public class IndexMetaInfo {
|
|||
|
||||
private int length;
|
||||
|
||||
public IndexMetaInfo(String name, long offset, int length) {
|
||||
this.fileName = name;
|
||||
this.offset = offset;
|
||||
this.length = length;
|
||||
}
|
||||
|
||||
private long startTime;
|
||||
|
||||
public String getFileName() {
|
||||
return fileName;
|
||||
|
|
@ -29,4 +22,8 @@ public class IndexMetaInfo {
|
|||
public int getLength() {
|
||||
return length;
|
||||
}
|
||||
|
||||
public long getStartTime() {
|
||||
return startTime;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,17 +1,54 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/3.
|
||||
*/
|
||||
import static com.a.eye.skywalking.storage.config.Config.DataIndex.TABLE_NAME;
|
||||
|
||||
public class IndexOperator {
|
||||
|
||||
public List<IndexMetaInfo> find(String traceId) {
|
||||
return null;
|
||||
}
|
||||
private Connection connection;
|
||||
private long timestamp;
|
||||
|
||||
public void update(IndexMetaInfo meta) {
|
||||
private IndexOperator(long timestamp) {
|
||||
|
||||
}
|
||||
|
||||
public List<IndexMetaInfo> find(String taceId) {
|
||||
return new ArrayList<IndexMetaInfo>();
|
||||
}
|
||||
|
||||
|
||||
public void update(List<IndexMetaInfo> metaInfo) {
|
||||
|
||||
}
|
||||
|
||||
private Connection getConnection() {
|
||||
return connection;
|
||||
}
|
||||
|
||||
public static class Builder {
|
||||
private IndexOperator operator;
|
||||
private IndexOperatorHelper indexOperatorHelper;
|
||||
|
||||
private Builder(long timestamp) {
|
||||
operator = new IndexOperator(timestamp);
|
||||
indexOperatorHelper = new IndexOperatorHelper(operator.getConnection());
|
||||
}
|
||||
|
||||
public static Builder newBuilder(long timestamp) {
|
||||
return new Builder(timestamp);
|
||||
}
|
||||
|
||||
public IndexOperator build() {
|
||||
if (indexOperatorHelper.validateIsReady(TABLE_NAME)) {
|
||||
indexOperatorHelper.maintain();
|
||||
}
|
||||
|
||||
return operator;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,65 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.concurrent.locks.ReadWriteLock;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
|
||||
public class IndexOperatorCache {
|
||||
|
||||
private static final int MAX_CACHE_SIZE = 5;
|
||||
|
||||
private LRUCache<Long, IndexOperator> cachedOperators;
|
||||
|
||||
public IndexOperatorCache() {
|
||||
cachedOperators = new LRUCache<Long, IndexOperator>(MAX_CACHE_SIZE);
|
||||
}
|
||||
|
||||
public IndexOperator get(long timestamp) {
|
||||
return cachedOperators.get(timestamp);
|
||||
}
|
||||
|
||||
public void updateCache(long timestamp, IndexOperator operator) {
|
||||
cachedOperators.put(timestamp, operator);
|
||||
}
|
||||
|
||||
private static class LRUCache<K, V> {
|
||||
private final int MAX_CACHE_SIZE;
|
||||
private final float DEFAULT_LOAD_FACTOR = 0.75f;
|
||||
private LinkedHashMap<K, V> map;
|
||||
private ReadWriteLock cacheLock = new ReentrantReadWriteLock();
|
||||
|
||||
public LRUCache(int cacheSize) {
|
||||
MAX_CACHE_SIZE = cacheSize;
|
||||
int capacity = (int) Math.ceil(MAX_CACHE_SIZE / DEFAULT_LOAD_FACTOR) + 1;
|
||||
map = new LinkedHashMap(capacity, DEFAULT_LOAD_FACTOR, true) {
|
||||
@Override
|
||||
protected boolean removeEldestEntry(Map.Entry eldest) {
|
||||
return size() > MAX_CACHE_SIZE;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
public void put(K key, V value) {
|
||||
Lock lock = cacheLock.writeLock();
|
||||
try {
|
||||
lock.lock();
|
||||
map.put(key, value);
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public V get(K key) {
|
||||
Lock lock = cacheLock.readLock();
|
||||
try {
|
||||
lock.lock();
|
||||
return map.get(key);
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -1,8 +1,20 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/4.
|
||||
*/
|
||||
public class IndexOperatorFactory {
|
||||
|
||||
public static IndexOperator get(long timestamp) {
|
||||
return null;
|
||||
private static IndexOperatorCache operatorCache;
|
||||
|
||||
public static IndexOperator getIndexOperator(long timestamp) {
|
||||
IndexOperator operator = operatorCache.get(timestamp);
|
||||
|
||||
if (operator == null) {
|
||||
operator = IndexOperator.Builder.newBuilder(timestamp).build();
|
||||
operatorCache.updateCache(timestamp, operator);
|
||||
}
|
||||
|
||||
return operator;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,27 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
import java.sql.Connection;
|
||||
|
||||
public class IndexOperatorHelper {
|
||||
|
||||
public IndexOperatorHelper(Connection connection) {
|
||||
|
||||
}
|
||||
|
||||
public boolean validateIsReady(String tableName) {
|
||||
return false;
|
||||
}
|
||||
|
||||
public void maintain() {
|
||||
createTable();
|
||||
createIndex();
|
||||
}
|
||||
|
||||
private void createIndex() {
|
||||
|
||||
}
|
||||
|
||||
private void createTable() {
|
||||
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue