diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Config.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Config.java index ea20dd76a..a3183bfd4 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Config.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Config.java @@ -20,6 +20,11 @@ public class Config { public static class DataIndex { + public static String TABLE_NAME = "data_index"; + + public static String BASE_PATH = ""; + + public static String STORAGE_INDEX_FILE_NAME = ""; } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java new file mode 100644 index 000000000..96e7a0276 --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Constants.java @@ -0,0 +1,25 @@ +package com.a.eye.skywalking.storage.config; + +import static com.a.eye.skywalking.storage.config.Config.DataIndex.TABLE_NAME; + +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" + + " parent_level_id VARCHAR(1024) NOT NULL,\n" + + " level_id INT NOT NULL,\n" + + " file_name VARCHAR(10) 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,parent_level_id,level_id," + + "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 + "';"; + } +} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataConsumer.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataConsumer.java index 7b6758d78..c5034e2dd 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataConsumer.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataConsumer.java @@ -15,6 +15,11 @@ public class SpanDataConsumer implements IConsumer { private IndexDBConnectorCache cache; private DataFileWriter fileWriter; + public SpanDataConsumer(){ + cache = new IndexDBConnectorCache(); + fileWriter = new DataFileWriter(); + } + @Override public void consume(List data) { diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/ConnectorInitializeFailedException.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/ConnectorInitializeFailedException.java new file mode 100644 index 000000000..3d276aed3 --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/ConnectorInitializeFailedException.java @@ -0,0 +1,10 @@ +package com.a.eye.skywalking.storage.data.exception; + +/** + * Created by xin on 2016/11/5. + */ +public class ConnectorInitializeFailedException extends RuntimeException { + public ConnectorInitializeFailedException(String message, Exception e) { + super(message, e); + } +} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/DataFileOperatorCreateFailedException.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/DataFileOperatorCreateFailedException.java new file mode 100644 index 000000000..d406f6cfc --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/DataFileOperatorCreateFailedException.java @@ -0,0 +1,10 @@ +package com.a.eye.skywalking.storage.data.exception; + +/** + * Created by xin on 2016/11/5. + */ +public class DataFileOperatorCreateFailedException extends RuntimeException { + public DataFileOperatorCreateFailedException(String message, Exception e){ + super(message, e); + } +} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/IndexMetaPersistenceFailedException.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/IndexMetaPersistenceFailedException.java new file mode 100644 index 000000000..17acbaf89 --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/IndexMetaPersistenceFailedException.java @@ -0,0 +1,7 @@ +package com.a.eye.skywalking.storage.data.exception; + +public class IndexMetaPersistenceFailedException extends RuntimeException { + public IndexMetaPersistenceFailedException(String message, Exception e){ + super(message, e); + } +} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/SpanDataPersistenceFailedException.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/SpanDataPersistenceFailedException.java new file mode 100644 index 000000000..07956c34e --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/SpanDataPersistenceFailedException.java @@ -0,0 +1,9 @@ +package com.a.eye.skywalking.storage.data.exception; + +/** + * Created by xin on 2016/11/4. + */ +public class SpanDataPersistenceFailedException extends RuntimeException { + public SpanDataPersistenceFailedException(Exception e) { + } +} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java index fd642d613..1ce3f94c1 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java @@ -1,10 +1,94 @@ package com.a.eye.skywalking.storage.data.file; +import com.a.eye.skywalking.storage.config.Config; +import com.a.eye.skywalking.storage.data.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.index.IndexMetaInfo; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; + /** * 数据文件 */ public class DataFile { + + private String fileName; + private long currentOffset; + private DataFileOperator operator; + + public DataFile() { + this.fileName = System.currentTimeMillis() + ""; + this.currentOffset = 0; + operator = new DataFileOperator(); + } + + public DataFile(String fileName, long offset) { + this.fileName = fileName; + this.currentOffset = offset; + operator = new DataFileOperator(); + } + + public DataFile(File file) { + this.fileName = file.getName(); + this.currentOffset = file.length(); + operator = new DataFileOperator(); + } + public boolean overLimitLength() { - return false; + return currentOffset >= Config.DataFile.MAX_LENGTH; + } + + public IndexMetaInfo write(SpanData data) { + byte[] bytes = data.toByteArray(); + try { + operator.getWriter().write(bytes); + IndexMetaInfo metaInfo = new IndexMetaInfo(fileName, currentOffset, bytes.length); + currentOffset += bytes.length; + return metaInfo; + } catch (IOException e) { + throw new SpanDataPersistenceFailedException(e); + } + } + + public void flush() { + try { + operator.getWriter().flush(); + } catch (IOException e) { + throw new SpanDataPersistenceFailedException(e); + } + } + + class DataFileOperator { + private FileOutputStream writer; + private FileInputStream reader; + + public FileOutputStream getWriter() { + + if (writer == null) { + try { + writer = new FileOutputStream(new File(fileName)); + } catch (IOException e) { + throw new DataFileOperatorCreateFailedException("Failed to create datafile output stream", e); + } + } + + return writer; + } + + public FileInputStream getReader() { + if (reader == null) { + try { + reader = new FileInputStream(new File(fileName)); + } catch (IOException e) { + throw new DataFileOperatorCreateFailedException("Failed to create datafile input stream", e); + } + } + + return reader; + } } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileLoader.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileLoader.java index 9d5bbe50c..32973177d 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileLoader.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileLoader.java @@ -1,16 +1,24 @@ package com.a.eye.skywalking.storage.data.file; +import java.io.File; import java.util.ArrayList; import java.util.List; public class DataFileLoader { - public DataFileLoader(String basePath) { + private String basePath; + public DataFileLoader(String basePath) { + this.basePath = basePath; } public List load() { - return new ArrayList(); + File dataFileDir = new File(basePath); + List allDataFile = new ArrayList(); + for (File fileEntry : dataFileDir.listFiles()) { + allDataFile.add(new DataFile(fileEntry)); + } + return allDataFile; } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileOperatorFactory.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileOperatorFactory.java deleted file mode 100644 index 126ca5061..000000000 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileOperatorFactory.java +++ /dev/null @@ -1,10 +0,0 @@ -package com.a.eye.skywalking.storage.data.file; - -import com.a.eye.skywalking.storage.data.index.IndexMetaInfo; - - -public class DataFileOperatorFactory { - public static DataFileReader getDataFileReader(IndexMetaInfo info) { - return new DataFileReader(info); - } -} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileReader.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileReader.java deleted file mode 100644 index 67f611d9f..000000000 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileReader.java +++ /dev/null @@ -1,17 +0,0 @@ -package com.a.eye.skywalking.storage.data.file; - -import com.a.eye.skywalking.storage.data.index.IndexMetaInfo; - -/** - * Created by xin on 2016/11/4. - */ -public class DataFileReader { - - public DataFileReader(IndexMetaInfo info) { - - } - - public byte[] read() { - return new byte[0]; - } -} diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileWriter.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileWriter.java index 0eaf3bd68..4e6c34281 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileWriter.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileWriter.java @@ -14,11 +14,16 @@ public class DataFileWriter { } public IndexMetaCollections write(List spanData) { - if (dataFile.overLimitLength()) { dataFile = DataFilesManager.createNewDataFile(); } - return null; + IndexMetaCollections collections = new IndexMetaCollections(); + for (SpanData data : spanData) { + collections.add(dataFile.write(data)); + } + dataFile.flush(); + + return collections; } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/UnFinishedDataFilePicker.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/UnFinishedDataFilePicker.java index 21f1a0157..9df47927e 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/UnFinishedDataFilePicker.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/UnFinishedDataFilePicker.java @@ -11,10 +11,16 @@ public class UnFinishedDataFilePicker { private List dataFiles; public UnFinishedDataFilePicker(List dataFiles) { - + this.dataFiles = dataFiles; } public List pickUp() { - return new ArrayList(); + List result = new ArrayList(); + for (DataFile file : dataFiles) { + if (file.overLimitLength()){ + result.add(file); + } + } + return result; } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java index be1b56965..1a2e8f4da 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java @@ -1,30 +1,123 @@ package com.a.eye.skywalking.storage.data.index; +import com.a.eye.skywalking.storage.config.Config; +import com.a.eye.skywalking.storage.data.exception.ConnectorInitializeFailedException; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.sql.*; + +import static com.a.eye.skywalking.storage.config.Constants.SQL.*; + /** * Created by xin on 2016/11/4. */ public class IndexDBConnector { - private long timestamp; + private static final int MAX_BATCH_SIZE = 20; + + private static Logger logger = LogManager.getLogger(IndexDBConnector.class); + + static { + try { + Class.forName("org.hsqldb.jdbc.JDBCDriver"); + } catch (ClassNotFoundException e) { + //never + } + } + + private long timestamp; + private Connection connection; + private ConnectURLGenerator generator = + new ConnectURLGenerator(Config.DataIndex.BASE_PATH, Config.DataIndex.STORAGE_INDEX_FILE_NAME); public IndexDBConnector(long timestamp) { - + this.timestamp = timestamp; + createConnection(); + createTableAndIndexIfNecessary(); } - private void validate() { - + private void createTableAndIndexIfNecessary() { + try { + if (validateTableIsExists()) { + createTable(); + createIndex(); + } + } catch (SQLException e) { + throw new ConnectorInitializeFailedException("Failed to create table and index.", e); + } } - private void createTable() { + private void createConnection() { + try { + connection = DriverManager.getConnection(generator.generate(timestamp)); + connection.setAutoCommit(true); + } catch (SQLException e) { + throw new ConnectorInitializeFailedException("Failed to create connection.", e); + } } - private void createIndex() { + private boolean validateTableIsExists() throws SQLException { + PreparedStatement ps = connection.prepareStatement(QUERY_TABLES); + ResultSet rs = ps.executeQuery(); + rs.next(); + boolean exists = rs.getInt("TABLE_COUNT") == 1; + rs.close(); + ps.close(); + + return exists; } + private void createTable() throws SQLException { + PreparedStatement ps = connection.prepareStatement(CREATE_TABLE); + ps.execute(); + ps.close(); + } + + private void createIndex() throws SQLException { + PreparedStatement ps = connection.prepareStatement(CREATE_INDEX); + ps.execute(); + ps.close(); + } public long getTimestamp() { return timestamp; } + + public void batchUpdate(IndexMetaGroup metaGroup) throws SQLException { + int currentIndex = 0; + PreparedStatement ps = connection.prepareStatement(INSERT_INDEX); + for (IndexMetaInfo metaInfo : metaGroup.getMetaInfo()) { + ps.setString(1, metaInfo.getTraceId()); + ps.setString(2, metaInfo.getParentLevelId()); + ps.setInt(3, metaInfo.getLevelId()); + ps.setString(4, metaInfo.getFileName()); + ps.setLong(5, metaInfo.getOffset()); + ps.setInt(6, metaInfo.getLength()); + ps.addBatch(); + if (++currentIndex > MAX_BATCH_SIZE) { + ps.executeBatch(); + } + } + ps.executeBatch(); + ps.close(); + } + + class ConnectURLGenerator { + + private String basePath; + private String dbFileName; + + private ConnectURLGenerator(String basePath, String dbFileName) { + this.basePath = basePath; + this.dbFileName = dbFileName; + } + + + public String generate(long timestamp) { + return "jdbc:hsqldb:file:" + basePath + "/" + timestamp + "/" + dbFileName; + } + } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnectorCache.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnectorCache.java index a2474da4b..fc649d52f 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnectorCache.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnectorCache.java @@ -17,10 +17,15 @@ public class IndexDBConnectorCache { } public IndexDBConnector get(long timestamp) { - return cachedOperators.get(timestamp); + IndexDBConnector connector = cachedOperators.get(timestamp); + if (connector == null) { + connector = new IndexDBConnector(timestamp); + updateCache(timestamp, connector); + } + return connector; } - public void updateCache(long timestamp, IndexDBConnector operator) { + private void updateCache(long timestamp, IndexDBConnector operator) { cachedOperators.put(timestamp, operator); } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaCollections.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaCollections.java index c24e5ec5f..40c92f2dd 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaCollections.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaCollections.java @@ -11,7 +11,12 @@ import java.util.List; public class IndexMetaCollections { private List metaInfo; - private BlockFinder finder = BlockIndexEngine.newFinder(); + private BlockFinder finder; + + public IndexMetaCollections() { + metaInfo = new ArrayList<>(); + finder = BlockIndexEngine.newFinder(); + } public Iterator group() { List indexMetaGroups = new ArrayList(); @@ -35,4 +40,7 @@ public class IndexMetaCollections { } + public void add(IndexMetaInfo info) { + metaInfo.add(info); + } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaGroup.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaGroup.java index e791a5498..bb6386cd1 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaGroup.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaGroup.java @@ -1,6 +1,7 @@ package com.a.eye.skywalking.storage.data.index; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; /** diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java index b08f4528f..70d257362 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java @@ -11,6 +11,12 @@ public class IndexMetaInfo { private long startTime; + public IndexMetaInfo(String fileName, long offset, int length) { + this.fileName = fileName; + this.offset = offset; + this.length = length; + } + public String getFileName() { return fileName; } @@ -26,4 +32,16 @@ public class IndexMetaInfo { public long getStartTime() { return startTime; } + + public String getTraceId() { + return null; + } + + public String getParentLevelId() { + return null; + } + + public int getLevelId() { + return 0; + } } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperator.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperator.java index b2db058b8..a87bc6ae8 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperator.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperator.java @@ -1,5 +1,7 @@ package com.a.eye.skywalking.storage.data.index; +import com.a.eye.skywalking.storage.data.exception.IndexMetaPersistenceFailedException; + import java.util.ArrayList; import java.util.List; @@ -19,7 +21,11 @@ public class IndexOperator { public void batchUpdate(IndexMetaGroup metaGroup) { - + try { + connector.batchUpdate(metaGroup); + } catch (Exception e) { + throw new IndexMetaPersistenceFailedException("Failed to batch save index meta", e); + } } public static IndexOperator newOperator(IndexDBConnector indexDBConnector) { diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperatorHelper.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperatorHelper.java deleted file mode 100644 index 292b42a05..000000000 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperatorHelper.java +++ /dev/null @@ -1,27 +0,0 @@ -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() { - - } -}