diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockFinder.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockFinder.java index f812b997a..4dc21537d 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockFinder.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockFinder.java @@ -26,4 +26,9 @@ public class BlockFinder { return index; } + + public long findLastBlockIndex(){ + return l2Cache.getLastBlockIndex(); + } + } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockIndexEngine.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockIndexEngine.java index 85b0f9a96..386c45126 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockIndexEngine.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockIndexEngine.java @@ -6,7 +6,6 @@ public class BlockIndexEngine { private static L1Cache l1Cache; private static L2Cache l2Cache; - public static void start(){ l1Cache = new L1Cache(); l2Cache = new L2Cache(); diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/L2Cache.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/L2Cache.java index 9b60a0ee3..f3efa5af1 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/L2Cache.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/L2Cache.java @@ -40,4 +40,8 @@ public class L2Cache { lock.unlock(); } } + + public long getLastBlockIndex() { + return cacheData.last(); + } } 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 a3183bfd4..580fc3bfc 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 @@ -26,5 +26,7 @@ public class Config { public static String BASE_PATH = ""; public static String STORAGE_INDEX_FILE_NAME = ""; + + public static long MAX_CAPACITY_PER_INDEX = 1000 * 1000 * 1000 * 1000; } } 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 index 96e7a0276..322558054 100644 --- 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 @@ -21,5 +21,8 @@ public class Constants { public static final String QUERY_TABLES = "SELECT count(1) AS TABLE_COUNT FROM INFORMATION_SCHEMA.TABLES " + "WHERE TABLE_NAME= '" + TABLE_NAME + "';"; + + public static final String QUERY_INDEX_SIZE = "SELECT count(1) AS INDEX_SIZE FROM " + TABLE_NAME; } + } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java new file mode 100644 index 000000000..752991ac7 --- /dev/null +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java @@ -0,0 +1,73 @@ +package com.a.eye.skywalking.storage.data; + +import com.a.eye.skywalking.storage.block.index.BlockIndexEngine; +import com.a.eye.skywalking.storage.data.index.IndexDBConnector; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.sql.SQLException; +import java.util.concurrent.atomic.AtomicLong; + +import static com.a.eye.skywalking.storage.config.Config.DataIndex.MAX_CAPACITY_PER_INDEX; + +/** + * Created by xin on 2016/11/6. + */ +public class IndexDataCapacityMonitor { + + private static Logger logger = LogManager.getLogger(IndexDataCapacityMonitor.class); + + private static Detector detector; + + public static void addIndexData(long timestamp, int size) { + if (detector.isDetectFor(timestamp)) { + detector.add(size); + } + } + + private static class Detector { + + private AtomicLong currentSize; + private long timestamp; + + public Detector(long timestamp) { + this.timestamp = timestamp; + currentSize = new AtomicLong(); + } + + public Detector(long timestamp, long currentSize) { + this.currentSize = new AtomicLong(currentSize); + this.timestamp = timestamp; + } + + public boolean isDetectFor(long timestamp) { + return this.timestamp == timestamp; + } + + public void add(int updateRecordSize) { + if (currentSize.addAndGet(updateRecordSize) > MAX_CAPACITY_PER_INDEX * 0.8) { + notificationAddNewBlockIndex(); + } + } + } + + private static void notificationAddNewBlockIndex() { + long timestamp = System.currentTimeMillis() + 5 * 60 * 1000; + BlockIndexEngine.newUpdator().addRecord(timestamp); + detector = new Detector(timestamp); + } + + public static void start() { + long timestamp = BlockIndexEngine.newFinder().findLastBlockIndex(); + + IndexDBConnector dbConnector = new IndexDBConnector(timestamp); + long count = 0; + try { + count = dbConnector.fetchIndexSize(); + } catch (SQLException e) { + logger.error("Failed to to fetch index size from DB:{}", timestamp, e); + } + detector = new Detector(timestamp, count); + + } +} 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 1a2e8f4da..89c8e0a9a 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 @@ -63,7 +63,7 @@ public class IndexDBConnector { ResultSet rs = ps.executeQuery(); rs.next(); - boolean exists = rs.getInt("TABLE_COUNT") == 1; + boolean exists = rs.getInt("TABLE_COUNT") == 1; rs.close(); ps.close(); @@ -105,6 +105,18 @@ public class IndexDBConnector { ps.close(); } + public long fetchIndexSize() throws SQLException { + PreparedStatement ps = connection.prepareStatement(QUERY_INDEX_SIZE); + ResultSet rs = ps.executeQuery(); + rs.next(); + + long indexSize = rs.getLong("INDEX_SIZE"); + rs.close(); + ps.close(); + + return indexSize; + } + class ConnectURLGenerator { private String basePath; 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 bb6386cd1..1e4c90b2b 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,7 +1,6 @@ package com.a.eye.skywalking.storage.data.index; import java.util.ArrayList; -import java.util.Iterator; import java.util.List; /** @@ -47,4 +46,8 @@ public class IndexMetaGroup { public int hashCode() { return (int) (timestamp ^ (timestamp >>> 32)); } + + public int size() { + return metaInfo.size(); + } } 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 a87bc6ae8..279de6435 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,6 @@ package com.a.eye.skywalking.storage.data.index; +import com.a.eye.skywalking.storage.data.IndexDataCapacityMonitor; import com.a.eye.skywalking.storage.data.exception.IndexMetaPersistenceFailedException; import java.util.ArrayList; @@ -23,6 +24,7 @@ public class IndexOperator { public void batchUpdate(IndexMetaGroup metaGroup) { try { connector.batchUpdate(metaGroup); + IndexDataCapacityMonitor.addIndexData(timestamp, metaGroup.size()); } catch (Exception e) { throw new IndexMetaPersistenceFailedException("Failed to batch save index meta", e); }