diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java index 3667f8e66..0deba4c67 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java @@ -53,7 +53,11 @@ public class ChainBuildReducer extends Reducer { + callChainData, e); } } - container.saveToHBase(); - chainTree.saveToHbase(); + try { + container.saveToHBase(); + chainTree.saveToHbase(); + } catch (Exception e) { + logger.error("Failed to save summaryresult/chainTree.", e); + } } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/DBCallChainInfoDao.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/DBCallChainInfoDao.java index 6214d8747..8efb78187 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/DBCallChainInfoDao.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/DBCallChainInfoDao.java @@ -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 updateChainInfo) - throws SQLException { - PreparedStatement preparedStatement = null; - try { - preparedStatement = connection - .prepareStatement("UPDATE sw_chain_detail SET update_time = ? WHERE cid = ?"); - for (Map.Entry 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 { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainDetail.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainDetailForMysql.java similarity index 67% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainDetail.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainDetailForMysql.java index ed8ab9e66..69e2e3b21 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainDetail.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainDetailForMysql.java @@ -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 chainNodeMap = new HashMap(); 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 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 getChainNodes() { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java index 5fc868842..b2d2af1cb 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java @@ -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> hasBeenMergedChainIds; + private Map callChainDetailMap; + // 本次Reduce合并过的调用链 private Map combineChains; @@ -21,6 +25,7 @@ public class SpecificTimeCallTreeMergedChainIdContainer { this.treeToken = treeToken; hasBeenMergedChainIds = new HashMap>(); combineChains = new HashMap(); + callChainDetailMap = new HashMap(); } 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 entry : callChainDetailMap.entrySet()){ + entry.getValue().saveToMysql(); + } } /** * 保存被合并的cid信息列表 - * + * * @throws IOException * @throws InterruptedException */ @@ -69,7 +88,7 @@ public class SpecificTimeCallTreeMergedChainIdContainer { /** * 保存已经合并的调用链信息,包含调用链明细 - * + * * @throws IOException * @throws InterruptedException */