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

DataFinder获取连接改为从缓存中获取

This commit is contained in:
ascrutae 2016-11-13 09:32:47 +08:00
parent cdcb440281
commit 3de3dfdc3d
13 changed files with 322 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -46,6 +46,11 @@
<artifactId>skywalking-registry</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>2.4.3</version>
</dependency>
</dependencies>
<build>
<plugins>

11
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/Main.java
View File

@ -14,10 +14,13 @@ import com.a.eye.skywalking.storage.config.ConfigInitializer;
import com.a.eye.skywalking.storage.data.IndexDataCapacityMonitor;
import com.a.eye.skywalking.storage.notifier.SearchNotifier;
import com.a.eye.skywalking.storage.notifier.StorageNotifier;
import com.a.eye.skywalking.storage.util.NetUtils;
import java.io.IOException;
import java.util.Properties;
import static com.a.eye.skywalking.storage.config.Config.RegistryCenter.REGISTRY_PATH_PREFIX;
/**
* Created by xin on 2016/11/12.
*/
@ -34,13 +37,14 @@ public class Main {
public static void main(String[] args) {
try {
initializeParam();
new Thread(new IndexDataCapacityMonitor()).start();
transferService =
TransferServiceBuilder.newBuilder(Config.Server.PORT).startSpanStorageService(new StorageNotifier())
.startTraceSearchService(new SearchNotifier()).build();
transferService.start();
logger.info("transfer service started successfully!");
new Thread(new IndexDataCapacityMonitor()).start();
registryNode();
logger.info("storage service started successfully!");
Thread.currentThread().join();
@ -52,12 +56,15 @@ public class Main {
}
private static void registryNode() {
//TODO auth info auth schema
RegistryCenter registryCenter =
RegistryCenterFactory.INSTANCE.getRegistryCenter(CenterType.DEFAULT_CENTER_TYPE);
Properties registerConfig = new Properties();
registerConfig.setProperty(ZookeeperConfig.CONNECT_URL, Config.RegistryCenter.CONNECT_URL);
registerConfig.setProperty(ZookeeperConfig.AUTH_SCHEMA, Config.RegistryCenter.AUTH_SCHEMA);
registerConfig.setProperty(ZookeeperConfig.AUTH_INFO, Config.RegistryCenter.AUTH_INFO);
registryCenter.start(registerConfig);
registryCenter.register(
REGISTRY_PATH_PREFIX + NetUtils.getLocalAddress().getHostAddress() + ":" + Config.Server.PORT);
}
private static void initializeParam() throws IllegalAccessException, IOException {

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

@ -33,10 +33,23 @@ public class Config {
public static String STORAGE_INDEX_FILE_NAME = "dataIndex";
public static long MAX_CAPACITY_PER_INDEX = 1000 * 1000 * 1000 * 1000;
}
public static class RegistryCenter {
public static String AUTH_INFO = "";
public static String AUTH_SCHEMA = "";
public static String CONNECT_URL = "127.0.0.1:2181";
public static String REGISTRY_PATH_PREFIX = "/storage_list/";
}
public static class SpanFinder {
public static int MAX_CACHE_SIZE = 10;
}
}

68
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataFinder.java
View File

@ -1,22 +1,31 @@
package com.a.eye.skywalking.storage.data;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.storage.block.index.BlockIndexEngine;
import com.a.eye.skywalking.storage.config.Config;
import com.a.eye.skywalking.storage.data.file.DataFileReader;
import com.a.eye.skywalking.storage.data.index.*;
import com.a.eye.skywalking.storage.data.spandata.SpanData;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.sql.SQLException;
import java.util.*;
public class SpanDataFinder {
private static ILog logger = LogManager.getLogger(SpanDataFinder.class);
private static IndexDataSourceCache datasourceCache = new IndexDataSourceCache(Config.SpanFinder.MAX_CACHE_SIZE);
public static List<SpanData> find(String traceId) {
long blockIndex = BlockIndexEngine.newFinder().find(fetchStartTimeFromTraceId(traceId));
if (blockIndex == 0) {
return new ArrayList<SpanData>();
}
IndexDBConnector indexDBConnector = new IndexDBConnector(blockIndex);
IndexDBConnector indexDBConnector = fetchIndexDBConnector(blockIndex);
IndexMetaCollection indexMetaCollection = indexDBConnector.queryByTraceId(traceId);
indexDBConnector.close();
Iterator<IndexMetaGroup<String>> iterator =
IndexMetaCollections.group(indexMetaCollection, new GroupKeyBuilder<String>() {
@ -35,8 +44,59 @@ public class SpanDataFinder {
return result;
}
private static IndexDBConnector fetchIndexDBConnector(long blockIndex) {
HikariDataSource datasource = getOrCreate(blockIndex);
IndexDBConnector indexDBConnector = null;
try {
indexDBConnector = new IndexDBConnector(datasource.getConnection());
} catch (SQLException e) {
logger.warn("Failed to get connection from datasource,", e);
indexDBConnector = new IndexDBConnector(blockIndex);
}
return indexDBConnector;
}
private static HikariDataSource getOrCreate(long blockIndex) {
HikariDataSource datasource = datasourceCache.get(blockIndex);
if (datasource == null) {
HikariConfig dataSourceConfig = generateDatasourceConfig(blockIndex);
datasource = new HikariDataSource(dataSourceConfig);
datasourceCache.put(blockIndex, datasource);
}
return datasource;
}
private static HikariConfig generateDatasourceConfig(long blockIndex) {
HikariConfig config = new HikariConfig();
config.setJdbcUrl(new ConnectURLGenerator(Config.DataIndex.BASE_PATH, Config.DataIndex.STORAGE_INDEX_FILE_NAME)
.generate(blockIndex));
config.setDriverClassName("org.hsqldb.jdbc.JDBCDriver");
config.setMaximumPoolSize(20);
config.setMinimumIdle(5);
return config;
}
private static long fetchStartTimeFromTraceId(String traceId) {
String[] traceIdSegment = traceId.split("\\.");
return Long.parseLong(traceIdSegment[traceIdSegment.length - 5]);
}
private static class IndexDataSourceCache extends LinkedHashMap<Long, HikariDataSource> {
private int cacheSize;
@Override
protected boolean removeEldestEntry(Map.Entry<Long, HikariDataSource> eldest) {
boolean removed = size() > cacheSize;
if (removed) {
eldest.getValue().close();
}
return removed;
}
public IndexDataSourceCache(int cacheSize) {
super((int) Math.ceil(cacheSize / 0.75) + 1, 0.75f, true);
this.cacheSize = cacheSize;
}
}
}

20
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/ConnectURLGenerator.java Normal file
View File

@ -0,0 +1,20 @@
package com.a.eye.skywalking.storage.data.index;
/**
* Created by xin on 2016/11/13.
*/
public class ConnectURLGenerator {
private String basePath;
private String dbFileName;
public ConnectURLGenerator(String basePath, String dbFileName) {
this.basePath = basePath;
this.dbFileName = dbFileName;
}
public String generate(long timestamp) {
return "jdbc:hsqldb:file:" + basePath + "/" + timestamp + "/" + dbFileName;
}
}

21
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java
View File

@ -41,6 +41,11 @@ public class IndexDBConnector {
createTableAndIndexIfNecessary();
}
public IndexDBConnector(Connection connection){
this.connection = connection;
createTableAndIndexIfNecessary();
}
private void createTableAndIndexIfNecessary() {
try {
if (validateTableIsExists()) {
@ -146,22 +151,6 @@ public class IndexDBConnector {
}
}
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;
}
}
public void close() {
try {
connection.close();

2
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnectorCache.java
View File

@ -41,7 +41,7 @@ public class IndexDBConnectorCache {
}
public LRUCache(int cacheSize) {
super((int) Math.ceil(MAX_CACHE_SIZE / 0.75) + 1, 0.75f, true);
super((int) Math.ceil(cacheSize / 0.75) + 1, 0.75f, true);
}
}
}

11
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/spandata/AckSpanData.java
View File

@ -40,4 +40,15 @@ public class AckSpanData extends AbstractSpanData {
return buildLevelId(ackSpan.getParentLevel(), ackSpan.getLevelId());
}
public long getCost() {
return ackSpan.getCost();
}
public String getExceptionStack() {
return ackSpan.getExceptionStack();
}
public int getStatusCode() {
return ackSpan.getStatusCode();
}
}

28
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/spandata/RequestSpanData.java
View File

@ -39,4 +39,32 @@ public class RequestSpanData extends AbstractSpanData {
public String getLevelId() {
return buildLevelId(requestSpan.getParentLevel(), requestSpan.getLevelId());
}
public String getAddress(){
return requestSpan.getAddress();
}
public String getApplicationId(){
return requestSpan.getApplicationId();
}
public String getProcessNo(){
return requestSpan.getProcessNo();
}
public long getStartTime(){
return requestSpan.getStartDate();
}
public String getBusinessKey() {
return requestSpan.getBussinessKey();
}
public String getCallType() {
return requestSpan.getCallType();
}
public int getType() {
return requestSpan.getSpanType();
}
}

38
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/notifier/SearchNotifier.java
View File

@ -3,50 +3,16 @@ package com.a.eye.skywalking.storage.notifier;
import com.a.eye.skywalking.network.grpc.Span;
import com.a.eye.skywalking.network.listener.TraceSearchNotifier;
import com.a.eye.skywalking.storage.data.SpanDataFinder;
import com.a.eye.skywalking.storage.data.spandata.AckSpanData;
import com.a.eye.skywalking.storage.data.spandata.RequestSpanData;
import com.a.eye.skywalking.storage.data.spandata.SpanData;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class SearchNotifier implements TraceSearchNotifier {
@Override
public List<Span> search(String s) {
List<SpanData> data = SpanDataFinder.find(s);
return mergeSpanData(data);
SpanDataHelper helper = new SpanDataHelper(data);
return helper.category().mergeData();
}
private List<Span> mergeSpanData(List<SpanData> data) {
//// TODO: 2016/11/12 需要修改
Map<String, RequestSpanData> requestSpen = new HashMap<String, RequestSpanData>();
Map<String, AckSpanData> ackSpen = new HashMap<String, AckSpanData>();
for (SpanData spanData : data) {
if (spanData instanceof RequestSpanData) {
requestSpen.put(spanData.getLevelId(), (RequestSpanData) spanData);
} else {
ackSpen.put(spanData.getLevelId(), (AckSpanData) spanData);
}
}
List<Span> mergedSpan = new ArrayList<Span>();
for (Map.Entry<String, RequestSpanData> entry : requestSpen.entrySet()) {
AckSpanData ackSpanData = ackSpen.get(entry.getKey());
if (ackSpanData != null) {
mergedSpan.add(mergeSpan(entry.getValue(), ackSpanData));
}
}
return mergedSpan;
}
private Span mergeSpan(RequestSpanData value, AckSpanData ackSpanData) {
return null;
}
}

64
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/notifier/SpanDataHelper.java Normal file
View File

@ -0,0 +1,64 @@
package com.a.eye.skywalking.storage.notifier;
import com.a.eye.skywalking.network.grpc.Span;
import com.a.eye.skywalking.storage.data.spandata.AckSpanData;
import com.a.eye.skywalking.storage.data.spandata.RequestSpanData;
import com.a.eye.skywalking.storage.data.spandata.SpanData;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Created by xin on 2016/11/12.
*/
public class SpanDataHelper {
public HashMap<String, RequestSpanData> levelIdRequestSpanDataMapping = new HashMap<String, RequestSpanData>();
public HashMap<String, AckSpanData> levelIdAckSpanDataMapping = new HashMap<String, AckSpanData>();
private List<SpanData> data;
public SpanDataHelper(List<SpanData> data) {
this.data = data;
}
public SpanDataHelper category() {
for (SpanData spanData : data) {
if (spanData instanceof RequestSpanData) {
levelIdRequestSpanDataMapping.put(spanData.getLevelId(), (RequestSpanData) spanData);
} else {
levelIdAckSpanDataMapping.put(spanData.getLevelId(), (AckSpanData) spanData);
}
}
return this;
}
public List<Span> mergeData() {
List<Span> span = new ArrayList<Span>();
for (Map.Entry<String, RequestSpanData> entry : levelIdRequestSpanDataMapping.entrySet()) {
AckSpanData ackSpanData = levelIdAckSpanDataMapping.get(entry.getKey());
if (ackSpanData != null) {
span.add(mergeSpan(entry.getValue(), ackSpanData));
}
}
return span;
}
private Span mergeSpan(RequestSpanData requestSpanData, AckSpanData ackSpanData) {
Span.Builder builder = Span.newBuilder().setAddress(requestSpanData.getAddress())
.setApplicationId(requestSpanData.getApplicationId()).setBusinessKey(requestSpanData.getBusinessKey())
.setCallType(requestSpanData.getCallType()).setCost(ackSpanData.getCost());
if (ackSpanData.getExceptionStack() != null && ackSpanData.getExceptionStack().length() > 0) {
builder = builder.setExceptionStack(ackSpanData.getExceptionStack());
}
builder = builder.setLevelId(requestSpanData.getLevelId()).setProcessNo(requestSpanData.getProcessNo())
.setSpanType(requestSpanData.getType()).setStarttime(requestSpanData.getStartTime())
.setStatusCode(ackSpanData.getStatusCode()).setTraceId(requestSpanData.getTraceId());
return builder.build();
}
}

74
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/util/NetUtils.java Normal file
View File

@ -0,0 +1,74 @@
package com.a.eye.skywalking.storage.util;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.Enumeration;
import java.util.regex.Pattern;
/**
* Created by xin on 2016/11/12.
*/
public class NetUtils {
private static ILog logger = LogManager.getLogger(NetUtils.class);
public static final String LOCALHOST = "127.0.0.1";
public static final String ANYHOST = "0.0.0.0";
private static final Pattern IP_PATTERN = Pattern.compile("\\d{1,3}(\\.\\d{1,3}){3,5}$");
public static InetAddress getLocalAddress() {
InetAddress localAddress = null;
try {
localAddress = InetAddress.getLocalHost();
if (isValidAddress(localAddress)) {
return localAddress;
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
try {
// 获取所有的网卡
Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
if (interfaces != null) {
while (interfaces.hasMoreElements()) {
try {
NetworkInterface network = interfaces.nextElement();
// 遍历网卡中所有绑定的地址
Enumeration<InetAddress> addresses = network.getInetAddresses();
if (addresses != null) {
while (addresses.hasMoreElements()) {
try {
InetAddress address = addresses.nextElement();
// 判断地址是否为合法的IP地址
if (isValidAddress(address)) {
return address;
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
}
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
}
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
return localAddress;
}
private static boolean isValidAddress(InetAddress address) {
if (address == null || address.isLoopbackAddress())
return false;
String name = address.getHostAddress();
// 不能是0.0.0.0 也不能是127.0.0.1 并且还得符合IP的正则
return (name != null && !ANYHOST.equals(name) && !LOCALHOST.equals(name) && IP_PATTERN.matcher(name).matches());
}
}

38
skywalking-storage-center/skywalking-storage/src/main/resources/config.properties
View File

@ -1,21 +1,35 @@
#
server.port = 34000
server.port=34000
#
blockindex.storage_base_path= /tmp/skywalking/block-index
#
blockindex.data_file_index_file_name= data_file.index
#
datafile.base_path= /tmp/skywalking/data/file
blockindex.storage_base_path=/tmp/skywalking/block-index
#
blockindex.data_file_index_file_name=data_file.index
#
#
#
datafile.base_path=/tmp/skywalking/data/file
#
datafile.max_length=3221225472
#
#
datafile.max_length= 3221225472
#存放数据文件索引表名
dataindex.table_name= data_index
dataindex.table_name=data_index
#数据文件索引存储位置
dataindex.base_path= /tmp/skywalking/data/index
dataindex.base_path=/tmp/skywalking/data/index
#
dataindex.storage_index_file_name= dataIndex
dataindex.storage_index_file_name=dataIndex
#
dataindex.max_capacity_per_index=1000000000
#
#
#
registrycenter.auth_info=
#
registrycenter.auth_schema=
#
registrycenter.connect_url=
#
registrycenter.registry_path_prefix=
#
dataindex.max_capacity_per_index= 1000000000