1.完成CallChain的数据存入Mysql表中
This commit is contained in:
zhang.xin
parent
8365964493
commit
97adb36320
|
|
@ -53,7 +53,11 @@ public class ChainBuildReducer extends Reducer<Text, Text, Text, IntWritable> {
|
|||
+ callChainData, e);
|
||||
}
|
||||
}
|
||||
container.saveToHBase();
|
||||
chainTree.saveToHbase();
|
||||
try {
|
||||
container.saveToHBase();
|
||||
chainTree.saveToHbase();
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to save summaryresult/chainTree.", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild;
|
||||
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainDetail;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainDetailForMysql;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainNode;
|
||||
import com.ai.cloud.skywalking.analysis.config.Config;
|
||||
import org.slf4j.Logger;
|
||||
|
|
@ -30,16 +30,16 @@ public class DBCallChainInfoDao {
|
|||
}
|
||||
}
|
||||
|
||||
public synchronized static void saveChainDetail(CallChainDetail callChainDetail)
|
||||
public synchronized static void saveChainDetail(CallChainDetailForMysql callChainDetailForMysql)
|
||||
throws SQLException {
|
||||
PreparedStatement preparedStatement = null;
|
||||
try {
|
||||
preparedStatement = connection
|
||||
.prepareStatement("INSERT INTO sw_chain_detail(cid,uid,traceLevelId,viewpoint,create_time)"
|
||||
+ " VALUES(?,?,?,?,?)");
|
||||
for (ChainNode chainNode : callChainDetail.getChainNodes()) {
|
||||
preparedStatement.setString(1, callChainDetail.getChainToken());
|
||||
preparedStatement.setString(2, callChainDetail.getUserId());
|
||||
for (ChainNode chainNode : callChainDetailForMysql.getChainNodes()) {
|
||||
preparedStatement.setString(1, callChainDetailForMysql.getChainToken());
|
||||
preparedStatement.setString(2, callChainDetailForMysql.getUserId());
|
||||
preparedStatement.setString(3, chainNode.getTraceLevelId());
|
||||
preparedStatement.setString(4, chainNode.getViewPoint() + ":"
|
||||
+ chainNode.getBusinessKey());
|
||||
|
|
@ -51,32 +51,7 @@ public class DBCallChainInfoDao {
|
|||
for (int i : result) {
|
||||
if (i != 1) {
|
||||
logger.error("Failed to save chain detail ["
|
||||
+ callChainDetail.getChainToken() + "]");
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (preparedStatement != null)
|
||||
preparedStatement.close();
|
||||
}
|
||||
connection.commit();
|
||||
}
|
||||
|
||||
public synchronized static void updateChainLastActiveTime(Map<String, Timestamp> updateChainInfo)
|
||||
throws SQLException {
|
||||
PreparedStatement preparedStatement = null;
|
||||
try {
|
||||
preparedStatement = connection
|
||||
.prepareStatement("UPDATE sw_chain_detail SET update_time = ? WHERE cid = ?");
|
||||
for (Map.Entry<String, Timestamp> entry : updateChainInfo
|
||||
.entrySet()) {
|
||||
preparedStatement.setTimestamp(1, entry.getValue());
|
||||
preparedStatement.setString(2, entry.getKey());
|
||||
preparedStatement.addBatch();
|
||||
}
|
||||
int[] result = preparedStatement.executeBatch();
|
||||
for (int i : result) {
|
||||
if (i != 1) {
|
||||
logger.error("Failed to update chain detail");
|
||||
+ callChainDetailForMysql.getChainToken() + "]");
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
|
|
|
|||
|
|
@ -12,20 +12,17 @@ import java.util.Collection;
|
|||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
public class CallChainDetail {
|
||||
private boolean isNormal = true;
|
||||
public class CallChainDetailForMysql {
|
||||
private String chainToken;
|
||||
private Map<String, ChainNode> chainNodeMap = new HashMap<String, ChainNode>();
|
||||
private String userId;
|
||||
|
||||
public CallChainDetail(ChainInfo chainInfo, boolean isNormal) {
|
||||
public CallChainDetailForMysql(ChainInfo chainInfo) {
|
||||
chainToken = chainInfo.getCID();
|
||||
for (ChainNode chainNode : chainInfo.getNodes()) {
|
||||
chainNodeMap.put(chainNode.getTraceLevelId(), chainNode);
|
||||
}
|
||||
userId = chainInfo.getUserId();
|
||||
|
||||
this.isNormal = isNormal;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -33,14 +30,8 @@ public class CallChainDetail {
|
|||
return new Gson().toJson(this);
|
||||
}
|
||||
|
||||
public void save(Put put) throws SQLException {
|
||||
for (Map.Entry<String, ChainNode> entry : chainNodeMap.entrySet()){
|
||||
put.addColumn(HBaseTableMetaData.TABLE_CHAIN_DETAIL.COLUMN_FAMILY_NAME.getBytes(),entry.getKey().getBytes(),
|
||||
entry.getValue().toString().getBytes());
|
||||
}
|
||||
if (isNormal) {
|
||||
DBCallChainInfoDao.saveChainDetail(this);
|
||||
}
|
||||
public void saveToMysql() throws SQLException {
|
||||
DBCallChainInfoDao.saveChainDetail(this);
|
||||
}
|
||||
|
||||
public Collection<ChainNode> getChainNodes() {
|
||||
|
|
@ -1,11 +1,13 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild.po;
|
||||
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainDetailForMysql;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.HBaseUtil;
|
||||
import com.ai.cloud.skywalking.analysis.config.HBaseTableMetaData;
|
||||
import com.google.gson.Gson;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.sql.SQLException;
|
||||
import java.util.*;
|
||||
|
||||
public class SpecificTimeCallTreeMergedChainIdContainer {
|
||||
|
|
@ -14,6 +16,8 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
|
|||
|
||||
private Map<String, List<String>> hasBeenMergedChainIds;
|
||||
|
||||
private Map<String, CallChainDetailForMysql> callChainDetailMap;
|
||||
|
||||
// 本次Reduce合并过的调用链
|
||||
private Map<String, ChainInfo> combineChains;
|
||||
|
||||
|
|
@ -21,6 +25,7 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
|
|||
this.treeToken = treeToken;
|
||||
hasBeenMergedChainIds = new HashMap<String, List<String>>();
|
||||
combineChains = new HashMap<String, ChainInfo>();
|
||||
callChainDetailMap = new HashMap<String, CallChainDetailForMysql>();
|
||||
}
|
||||
|
||||
public void addMergedChainIfNotContain(ChainInfo chainInfo) throws IOException {
|
||||
|
|
@ -34,6 +39,13 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
|
|||
if (!cIds.contains(chainInfo.getCID())) {
|
||||
cIds.add(chainInfo.getCID());
|
||||
combineChains.put(chainInfo.getCID(), chainInfo);
|
||||
|
||||
//
|
||||
if (chainInfo.getChainStatus() == ChainInfo.ChainStatus.NORMAL) {
|
||||
callChainDetailMap.put(chainInfo.getCID(), new CallChainDetailForMysql(chainInfo));
|
||||
}
|
||||
}else{
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -44,14 +56,21 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
|
|||
return treeToken + "@" + calendar.get(Calendar.YEAR) + "-" + calendar.get(Calendar.MONTH);
|
||||
}
|
||||
|
||||
public void saveToHBase() throws IOException, InterruptedException {
|
||||
public void saveToHBase() throws IOException, InterruptedException, SQLException {
|
||||
batchSaveCurrentHasBeenMergedChainInfo();
|
||||
batchSaveMergedChainId();
|
||||
batchSaveToMysql();
|
||||
}
|
||||
|
||||
private void batchSaveToMysql() throws SQLException {
|
||||
for (Map.Entry<String, CallChainDetailForMysql> entry : callChainDetailMap.entrySet()){
|
||||
entry.getValue().saveToMysql();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 保存被合并的cid信息列表
|
||||
*
|
||||
*
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
|
|
@ -69,7 +88,7 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
|
|||
|
||||
/**
|
||||
* 保存已经合并的调用链信息，包含调用链明细
|
||||
*
|
||||
*
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
|
|
|
|||
Loading…
Reference in New Issue