补充切换数据库逻辑
This commit is contained in:
ascrutae
parent
dd5cd7a1dc
commit
06f171d8af
|
|
@ -26,4 +26,9 @@ public class BlockFinder {
|
|||
return index;
|
||||
}
|
||||
|
||||
|
||||
public long findLastBlockIndex(){
|
||||
return l2Cache.getLastBlockIndex();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -40,4 +40,8 @@ public class L2Cache {
|
|||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public long getLastBlockIndex() {
|
||||
return cacheData.last();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue