diff --git a/skywalking-analysis/pom.xml b/skywalking-analysis/pom.xml index bd22c8694..891f02163 100644 --- a/skywalking-analysis/pom.xml +++ b/skywalking-analysis/pom.xml @@ -52,6 +52,11 @@ gson 2.2.2 + + mysql + mysql-connector-java + 5.1.38 + diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java index ef1a3dbbd..c13374e1e 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java @@ -3,6 +3,10 @@ package com.ai.cloud.skywalking.analysis.config; public class Config { public static class HBase { + public static String TABLE_CHAIN_DETAIL = "sw_chain_detail"; + + public static String TRACE_DETAIL_FAMILY_COLUMN = "chain_detail"; + public static String CHAIN_SUMMARY_COLUMN_FAMILY = "chain_summary"; public static String TRACE_INFO_COLUMN_FAMILY = "trace_info"; @@ -31,13 +35,13 @@ public class Config { public static class MySql { - public static String url; + public static String url = "jdbc:mysql://10.1.228.202:31316/test"; - public static String userName; + public static String userName = "devrdbusr21"; - public static String password; + public static String password = "devrdbusr21"; - public static String driverClass; + public static String driverClass = "com.mysql.jdbc.Driver"; } public static class Filter { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/dao/CallChainInfoDao.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/dao/CallChainInfoDao.java index 93ebc0759..390faf4ea 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/dao/CallChainInfoDao.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/dao/CallChainInfoDao.java @@ -1,11 +1,13 @@ package com.ai.cloud.skywalking.analysis.dao; import com.ai.cloud.skywalking.analysis.config.Config; -import com.ai.cloud.skywalking.analysis.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.model.ChainNode; +import com.ai.cloud.skywalking.analysis.reduce.ChainDetail; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.sql.*; +import java.util.Map; public class CallChainInfoDao { private static Logger logger = LoggerFactory.getLogger(CallChainInfoDao.class.getName()); @@ -25,14 +27,38 @@ public class CallChainInfoDao { } } - public static boolean existCallChainInfo(String cid, String userId) throws SQLException { - final String sql = "SELECT COUNT(cid) as TOTAL_SIZE FROM sw_chain_info WHERE cid = ? AND uid = ?"; - PreparedStatement ps = connection.prepareStatement(sql); - ps.setString(1, cid); - ps.setString(2, userId); - ResultSet resultSet = ps.executeQuery(); - resultSet.next(); - return resultSet.getInt("TOTAL_SIZE") > 0 ? true : false; + + public static void saveChainDetail(ChainDetail chainDetail) throws SQLException { + PreparedStatement preparedStatement = connection.prepareStatement("INSERT INTO sw_chain_detail(cid,uid,traceLevelId,viewpoint,create_time)" + + " VALUES(?,?,?,?,?)"); + for (ChainNode chainNode : chainDetail.getChainNodes()) { + preparedStatement.setString(1, chainDetail.getChainToken()); + preparedStatement.setString(2, chainDetail.getUserId()); + preparedStatement.setString(3, chainNode.getTraceLevelId()); + preparedStatement.setString(4, chainNode.getViewPoint() + ":" + chainNode.getBusinessKey()); + preparedStatement.setTimestamp(5, new Timestamp(System.currentTimeMillis())); + preparedStatement.addBatch(); + } + int[] result = preparedStatement.executeBatch(); + for (int i : result) { + //TODO + } + preparedStatement.close(); + connection.commit(); } + public static void updateChainDetail(Map updateChainInfo) throws SQLException { + PreparedStatement 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) { + //TODO + } + preparedStatement.close(); + connection.commit(); + } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java index f00ccf477..d02bb8103 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java @@ -1,8 +1,8 @@ package com.ai.cloud.skywalking.analysis.filter; import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.model.SpanEntry; +import com.ai.cloud.skywalking.analysis.mapper.CostMap; +import com.ai.cloud.skywalking.analysis.mapper.SpanEntry; public abstract class SpanNodeProcessFilter { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java index ebe6244d0..657a1b227 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java @@ -2,8 +2,8 @@ package com.ai.cloud.skywalking.analysis.filter.impl; import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.model.SpanEntry; +import com.ai.cloud.skywalking.analysis.mapper.CostMap; +import com.ai.cloud.skywalking.analysis.mapper.SpanEntry; public class AppendBusinessKeyFilter extends SpanNodeProcessFilter { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java index e8efbe3a7..ce4c58e6a 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java @@ -2,8 +2,8 @@ package com.ai.cloud.skywalking.analysis.filter.impl; import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.model.SpanEntry; +import com.ai.cloud.skywalking.analysis.mapper.CostMap; +import com.ai.cloud.skywalking.analysis.mapper.SpanEntry; public class CopyAttrFilter extends SpanNodeProcessFilter { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java index fcca43874..8a92857f5 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java @@ -2,8 +2,8 @@ package com.ai.cloud.skywalking.analysis.filter.impl; import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.model.SpanEntry; +import com.ai.cloud.skywalking.analysis.mapper.CostMap; +import com.ai.cloud.skywalking.analysis.mapper.SpanEntry; public class ProcessCostTimeFilter extends SpanNodeProcessFilter { @Override diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java index 527e7a1c2..8a1550565 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java @@ -2,8 +2,8 @@ package com.ai.cloud.skywalking.analysis.filter.impl; import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.model.SpanEntry; +import com.ai.cloud.skywalking.analysis.mapper.CostMap; +import com.ai.cloud.skywalking.analysis.mapper.SpanEntry; public class ReplaceAddressFilter extends SpanNodeProcessFilter { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java index a7aa1edfb..a4e133a74 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java @@ -2,8 +2,8 @@ package com.ai.cloud.skywalking.analysis.filter.impl; import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.model.SpanEntry; +import com.ai.cloud.skywalking.analysis.mapper.CostMap; +import com.ai.cloud.skywalking.analysis.mapper.SpanEntry; import com.ai.cloud.skywalking.analysis.util.TokenGenerator; public class TokenGenerateFilter extends SpanNodeProcessFilter { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java index 801c013b4..705248a20 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java @@ -4,8 +4,6 @@ import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessChain; import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; import com.ai.cloud.skywalking.analysis.model.ChainInfo; import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.model.SpanEntry; import com.ai.cloud.skywalking.analysis.util.HBaseUtil; import com.ai.cloud.skywalking.protocol.Span; import org.apache.hadoop.hbase.Cell; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CostMap.java similarity index 89% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CostMap.java index 55e54c871..484fb00f5 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CostMap.java @@ -1,4 +1,4 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.mapper; import java.util.HashMap; import java.util.Map; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/SpanEntry.java similarity index 97% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/SpanEntry.java index 7a564de9c..438cd86cd 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/SpanEntry.java @@ -1,5 +1,6 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.mapper; +import com.ai.cloud.skywalking.analysis.model.ChainNode; import com.ai.cloud.skywalking.protocol.CallType; import com.ai.cloud.skywalking.protocol.Span; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainDetail.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainDetail.java deleted file mode 100644 index 04e379c7f..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainDetail.java +++ /dev/null @@ -1,12 +0,0 @@ -package com.ai.cloud.skywalking.analysis.model; - -public class ChainDetail { - - public ChainDetail(ChainInfo chainInfo) { - - } - - public ChainDetail(UncategorizeChainInfo child) { - - } -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java index 927a44525..26ab6cec3 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java @@ -78,17 +78,13 @@ public class ChainInfo implements Writable { for (ChainNode node : nodes) { chainTokenDesc.append(node.getParentLevelId() + "." + node.getLevelId() + "-" + node.getNodeToken() + ";"); } - this.chainToken = TokenGenerator.generate(chainTokenDesc.toString()); + this.chainToken = TokenGenerator.generate(chainTokenDesc.toString()) + "-" + userId; } public ChainStatus getChainStatus() { return chainStatus; } - public void setChainStatus(ChainStatus chainStatus) { - this.chainStatus = chainStatus; - } - public void addNodes(ChainNode chainNode) { this.nodes.add(0, chainNode); if (chainNode.getStatus() == ChainNode.NodeStatus.ABNORMAL) { @@ -143,10 +139,6 @@ public class ChainInfo implements Writable { public void setChainToken(String chainToken) { this.chainToken = chainToken; } - - public long getStartDate() { - return startDate; - } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java index 9acdf2156..f75bb0e2c 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java @@ -1,19 +1,26 @@ package com.ai.cloud.skywalking.analysis.model; +import com.google.gson.GsonBuilder; +import com.google.gson.annotations.Expose; + public class ChainNode { - + @Expose private String nodeToken; - + @Expose private String viewPoint; + @Expose private String businessKey; + private long cost; private NodeStatus status; + @Expose private String parentLevelId; + @Expose private int levelId; + @Expose private String callType; private long startDate; - - // 不参与序列化 + @Expose private String userId; public String getNodeToken() { @@ -123,4 +130,9 @@ public class ChainNode { } } } + + @Override + public String toString() { + return new GsonBuilder().excludeFieldsWithoutExposeAnnotation().create().toJson(this); + } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CategorizedChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/CategorizedChainInfo.java similarity index 83% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CategorizedChainInfo.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/CategorizedChainInfo.java index 1c7c8f7e1..6c772d6ab 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CategorizedChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/CategorizedChainInfo.java @@ -1,7 +1,10 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.reduce; +import com.ai.cloud.skywalking.analysis.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.model.ChainNode; import com.google.gson.Gson; import com.google.gson.JsonObject; +import com.google.gson.JsonParser; import com.google.gson.reflect.TypeToken; import java.util.ArrayList; @@ -23,7 +26,7 @@ public class CategorizedChainInfo { if (flag) { stringBuilder.append(";"); } - stringBuilder.append(chainNode.getTraceLevelId() + "-" + chainNode.getViewPoint()); + stringBuilder.append((chainNode.getTraceLevelId() + "-" + chainNode.getNodeToken())); flag = true; } @@ -32,7 +35,7 @@ public class CategorizedChainInfo { } public CategorizedChainInfo(String value) { - JsonObject jsonObject = new Gson().fromJson(value, JsonObject.class); + JsonObject jsonObject = (JsonObject) new JsonParser().parse(value); chainToken = jsonObject.get("chainToken").getAsString(); chainStr = jsonObject.get("chainStr").getAsString(); children_Token = new Gson().fromJson(jsonObject.get("children_Token"), diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainDetail.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainDetail.java new file mode 100644 index 000000000..fd8fb2758 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainDetail.java @@ -0,0 +1,53 @@ +package com.ai.cloud.skywalking.analysis.reduce; + +import com.ai.cloud.skywalking.analysis.config.Config; +import com.ai.cloud.skywalking.analysis.dao.CallChainInfoDao; +import com.ai.cloud.skywalking.analysis.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.model.ChainNode; +import com.google.gson.Gson; +import org.apache.hadoop.hbase.client.Put; + +import java.sql.SQLException; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; + +public class ChainDetail { + private String chainToken; + private Map chainNodeMap = new HashMap(); + private String userId; + + public ChainDetail(ChainInfo chainInfo) { + chainToken = chainInfo.getChainToken(); + for (ChainNode chainNode : chainInfo.getNodes()) { + chainNodeMap.put(chainNode.getTraceLevelId(), chainNode); + } + userId = chainInfo.getUserId(); + } + + @Override + public String toString() { + return new Gson().toJson(this); + } + + public void save(Put put) throws SQLException { + for (Map.Entry entry : chainNodeMap.entrySet()){ + put.addColumn(Config.HBase.TRACE_DETAIL_FAMILY_COLUMN.getBytes(),entry.getKey().getBytes(), + entry.getValue().toString().getBytes()); + } + + CallChainInfoDao.saveChainDetail(this); + } + + public Collection getChainNodes() { + return chainNodeMap.values(); + } + + public String getUserId() { + return userId; + } + + public String getChainToken() { + return chainToken; + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java index ebb388ba6..10404dffd 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java @@ -1,8 +1,6 @@ package com.ai.cloud.skywalking.analysis.reduce; import com.ai.cloud.skywalking.analysis.model.ChainInfo; -import com.ai.cloud.skywalking.analysis.model.ChainRelate; -import com.ai.cloud.skywalking.analysis.model.Summary; import com.ai.cloud.skywalking.analysis.util.HBaseUtil; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.mapreduce.TableReducer; @@ -14,20 +12,27 @@ import java.util.Iterator; public class ChainInfoReduce extends TableReducer { @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { - ChainRelate chainRelate = HBaseUtil.selectCallChainRelationship(key.toString()); - Summary summary = new Summary(); - Iterator chainInfoIterator = values.iterator(); - while (chainInfoIterator.hasNext()) { - ChainInfo chainInfo = chainInfoIterator.next(); - chainRelate.addRelate(chainInfo); - summary.summary(chainInfo); + reduceAction(key.toString(), values.iterator()); + } + + public static void reduceAction(String key, Iterator chainInfoIterator) throws IOException, InterruptedException { + try { + ChainRelate chainRelate = HBaseUtil.selectCallChainRelationship(key.toString()); + Summary summary = new Summary(); + while (chainInfoIterator.hasNext()) { + ChainInfo chainInfo = chainInfoIterator.next(); + try { + chainRelate.addRelate(chainInfo); + summary.summary(chainInfo); + } catch (Exception e) { + continue; + } + } + + chainRelate.save(); + summary.save(); + } catch (Exception e) { + e.printStackTrace(); } - - chainRelate.save(); - // 入HBase库（关系表，Info表，汇总表） - summary.save(); - // 入Mysql表 - - } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNodeSpecificTimeWindowSummary.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainNodeSpecificTimeWindowSummary.java similarity index 70% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNodeSpecificTimeWindowSummary.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainNodeSpecificTimeWindowSummary.java index 4bb382c34..604fb7894 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNodeSpecificTimeWindowSummary.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainNodeSpecificTimeWindowSummary.java @@ -1,8 +1,10 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.reduce; import com.ai.cloud.skywalking.analysis.config.Config; +import com.ai.cloud.skywalking.analysis.model.ChainNode; import com.google.gson.Gson; import com.google.gson.JsonObject; +import com.google.gson.JsonParser; import com.google.gson.reflect.TypeToken; import java.util.HashMap; @@ -25,9 +27,9 @@ public class ChainNodeSpecificTimeWindowSummary { } public ChainNodeSpecificTimeWindowSummary(String value) { - JsonObject jsonObject = new Gson().fromJson(value, JsonObject.class); + JsonObject jsonObject = (JsonObject) new JsonParser().parse(value); traceLevelId = jsonObject.get("traceLevelId").getAsString(); - summerResultMap = new Gson().fromJson(jsonObject.get("summerResultMap").getAsString(), + summerResultMap = new Gson().fromJson(jsonObject.get("summerResultMap").toString(), new TypeToken>() { }.getType()); } @@ -36,23 +38,12 @@ public class ChainNodeSpecificTimeWindowSummary { return traceLevelId; } - public void setTraceLevelId(String traceLevelId) { - this.traceLevelId = traceLevelId; - } - - public Map getSummerResultMap() { - return summerResultMap; - } - - public void setSummerResultMap(Map summerResultMap) { - this.summerResultMap = summerResultMap; - } - public void summary(ChainNode node) { - SummaryResult summaryResult = summerResultMap.get(generateKey(node.getStartDate())); + String key = generateKey(node.getStartDate()); + SummaryResult summaryResult = summerResultMap.get(key); if (summaryResult == null) { summaryResult = new SummaryResult(); - summerResultMap.put(node.getTraceLevelId(), summaryResult); + summerResultMap.put(key, summaryResult); } summaryResult.summary(node); } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainRelate.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainRelate.java similarity index 81% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainRelate.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainRelate.java index 6d948ed17..139fe29b1 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainRelate.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainRelate.java @@ -1,12 +1,14 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.reduce; import com.ai.cloud.skywalking.analysis.config.Config; import com.ai.cloud.skywalking.analysis.config.Constants; +import com.ai.cloud.skywalking.analysis.model.ChainInfo; import com.ai.cloud.skywalking.analysis.util.HBaseUtil; 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 ChainRelate { @@ -37,20 +39,14 @@ public class ChainRelate { for (Map.Entry entry : categorizedChainInfoMap.entrySet()) { if (entry.getValue().isAlreadyContained(child)) { isContained = true; - }else if(entry.getValue().isContained(child)){ - entry.getValue().add(child); - chainDetailMap.put(child.getChainToken(), new ChainDetail(child)); + } else if (entry.getValue().isContained(child)) { + entry.getValue().add(child); isContained = true; } } if (!isContained) { uncategorizeChainInfoList.add(child); - - if (!uncategorizeChainInfoList.contains(child)) { - chainDetailMap.put(child.getChainToken(), new ChainDetail(child)); - } - } } @@ -75,7 +71,30 @@ public class ChainRelate { } } - public void save() { + public void save() throws SQLException { + saveChainRelationShip(); + saveChainDetail(); + } + + private void saveChainDetail() throws SQLException { + List puts = new ArrayList(); + for (Map.Entry entry : chainDetailMap.entrySet()) { + Put put1 = new Put(entry.getKey().getBytes()); + puts.add(put1); + entry.getValue().save(put1); + } + + + try { + HBaseUtil.saveChainDetails(puts); + } catch (IOException e) { + e.printStackTrace(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + + private void saveChainRelationShip() { Put put = new Put(getKey().getBytes()); put.addColumn(Config.HBase.CHAIN_RELATIONSHIP_COLUMN_FAMILY.getBytes(), Constants.UNCATEGORIZED_QUALIFIER_NAME.getBytes() @@ -94,10 +113,6 @@ public class ChainRelate { } } - public void addUncategorizeChain(UncategorizeChainInfo uncategorizeChainInfo) { - uncategorizeChainInfoList.add(uncategorizeChainInfo); - } - public void addCategorizeChain(String qualifierName, CategorizedChainInfo categorizedChainInfo) { categorizedChainInfoMap.put(qualifierName, categorizedChainInfo); } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainSpecificTimeWindowSummary.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainSpecificTimeWindowSummary.java similarity index 94% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainSpecificTimeWindowSummary.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainSpecificTimeWindowSummary.java index 2b444c6b7..b05abe0cd 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainSpecificTimeWindowSummary.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainSpecificTimeWindowSummary.java @@ -1,6 +1,7 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.reduce; import com.ai.cloud.skywalking.analysis.config.Config; +import com.ai.cloud.skywalking.analysis.model.ChainNode; import com.ai.cloud.skywalking.analysis.util.HBaseUtil; import org.apache.hadoop.hbase.client.Put; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/Summary.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/Summary.java similarity index 52% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/Summary.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/Summary.java index b6868b0ec..8198407d0 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/Summary.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/Summary.java @@ -1,20 +1,25 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.reduce; +import com.ai.cloud.skywalking.analysis.dao.CallChainInfoDao; +import com.ai.cloud.skywalking.analysis.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.model.ChainNode; import com.ai.cloud.skywalking.analysis.util.HBaseUtil; import org.apache.hadoop.hbase.client.Put; import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.sql.*; +import java.text.SimpleDateFormat; +import java.util.*; +import java.util.Date; public class Summary { private Map summaryResultMap; + private Map updateChainInfo; public Summary() { summaryResultMap = new HashMap(); + updateChainInfo = new HashMap(); } public void summary(ChainInfo chainInfo) { @@ -26,15 +31,27 @@ public class Summary { summaryResultMap.put(csk, chainSummaryResult); } - chainSummaryResult.summaryResult(node); + summaryResultMap.get(csk).summaryResult(node); } + + updateChainInfo.put(chainInfo.getChainToken(), new Timestamp(System.currentTimeMillis())); } private String generateChainSummaryKey(String chainToken, long startDate) { - return chainToken + "-" + (startDate / (1000 * 60 * 60)); + return chainToken + "-" + new SimpleDateFormat("yyyy/MM/dd HH:mm:ss"). + format(new Date(startDate / (1000 * 60 * 60) * (1000 * 60 * 60))); } - public void save() throws IOException, InterruptedException { + public void save() throws IOException, InterruptedException, SQLException { + batchSaveChainSpecificTimeWindowSummary(); + batchUpdateChainDetail(); + } + + private void batchUpdateChainDetail() throws SQLException { + CallChainInfoDao.updateChainDetail(updateChainInfo); + } + + private void batchSaveChainSpecificTimeWindowSummary() throws IOException, InterruptedException { List puts = new ArrayList(); for (Map.Entry entry : summaryResultMap.entrySet()) { Put put = new Put(entry.getKey().getBytes()); diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SummaryResult.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/SummaryResult.java similarity index 90% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SummaryResult.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/SummaryResult.java index 62bc2a848..8e96718ac 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SummaryResult.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/SummaryResult.java @@ -1,4 +1,6 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.reduce; + +import com.ai.cloud.skywalking.analysis.model.ChainNode; public class SummaryResult { private long totalCall; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/UncategorizeChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/UncategorizeChainInfo.java similarity index 50% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/UncategorizeChainInfo.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/UncategorizeChainInfo.java index e8dee0cd3..787236631 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/UncategorizeChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/UncategorizeChainInfo.java @@ -1,11 +1,18 @@ -package com.ai.cloud.skywalking.analysis.model; +package com.ai.cloud.skywalking.analysis.reduce; -import com.google.gson.Gson; +import com.ai.cloud.skywalking.analysis.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.model.ChainNode; +import com.google.gson.GsonBuilder; +import com.google.gson.annotations.Expose; public class UncategorizeChainInfo { + @Expose private String chainToken; + @Expose private String nodeRegEx; + private ChainInfo chainInfo; + public UncategorizeChainInfo() { } @@ -15,33 +22,33 @@ public class UncategorizeChainInfo { boolean flag = false; for (ChainNode node : chainInfo.getNodes()) { if (flag) { - stringBuilder.append("*"); + stringBuilder.append(";*"); } - stringBuilder.append(node.getTraceLevelId() + "-" + node.getViewPoint()); + stringBuilder.append((node.getTraceLevelId() + "-" + node.getNodeToken())); flag = true; } nodeRegEx = stringBuilder.toString(); + + this.chainInfo = chainInfo; } public String getChainToken() { return chainToken; } - public void setChainToken(String chainToken) { - this.chainToken = chainToken; - } - public String getNodeRegEx() { return nodeRegEx; } - public void setNodeRegEx(String nodeRegEx) { - this.nodeRegEx = nodeRegEx; + public ChainInfo getChainInfo() { + return chainInfo; } @Override public String toString() { - return new Gson().toJson(this); + GsonBuilder gsonBuilder = new GsonBuilder(); + gsonBuilder.excludeFieldsWithoutExposeAnnotation(); + return gsonBuilder.create().toJson(this); } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/ChainInfoUtil.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/ChainInfoUtil.java deleted file mode 100644 index 6a54d7d89..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/ChainInfoUtil.java +++ /dev/null @@ -1,16 +0,0 @@ -package com.ai.cloud.skywalking.analysis.util; - -import com.ai.cloud.skywalking.analysis.model.ChainInfo; - -public class ChainInfoUtil { - private ChainInfoUtil() { - // Non - } - - public static ChainInfo generateUncategorizedChainInfo() { - ChainInfo chainInfo = new ChainInfo(); - chainInfo.setChainToken("UNCATEGORIZED"); - chainInfo.setUserId("-1"); - return chainInfo; - } -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java index 66ad214c8..ffb7be20f 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java @@ -3,6 +3,7 @@ package com.ai.cloud.skywalking.analysis.util; import com.ai.cloud.skywalking.analysis.config.Config; import com.ai.cloud.skywalking.analysis.config.Constants; import com.ai.cloud.skywalking.analysis.model.*; +import com.ai.cloud.skywalking.analysis.reduce.*; import com.google.gson.Gson; import com.google.gson.reflect.TypeToken; import org.apache.hadoop.conf.Configuration; @@ -25,7 +26,7 @@ public class HBaseUtil { Table table = null; try { - table = connection.getTable(TableName.valueOf(Config.HBase.TRACE_INFO_TABLE_NAME)); + table = connection.getTable(TableName.valueOf(Config.HBase.TABLE_CHAIN_INFO)); } catch (IOException e) { logger.error("Cannot found table[" + Config.HBase.TRACE_INFO_TABLE_NAME + "]", e); } @@ -52,11 +53,14 @@ public class HBaseUtil { try { initHBaseClient(); // - createTableIfNeed(Config.HBase.TRACE_INFO_TABLE_NAME, Config.HBase.TRACE_INFO_COLUMN_FAMILY); + createTableIfNeed(Config.HBase.TABLE_CHAIN_INFO, Config.HBase.TRACE_INFO_COLUMN_FAMILY); // createTableIfNeed(Config.HBase.TABLE_CALL_CHAIN_RELATIONSHIP, Config.HBase.CHAIN_RELATIONSHIP_COLUMN_FAMILY); createTableIfNeed(Config.HBase.TABLE_CHAIN_SUMMARY, Config.HBase.CHAIN_SUMMARY_COLUMN_FAMILY); + + createTableIfNeed(Config.HBase.TABLE_CHAIN_DETAIL, Config.HBase.TRACE_DETAIL_FAMILY_COLUMN); + } catch (IOException e) { logger.error("Create table[{}] failed", Config.HBase.TRACE_INFO_TABLE_NAME, e); } @@ -114,12 +118,15 @@ public class HBaseUtil { public static ChainSpecificTimeWindowSummary selectChainSummaryResult(String key) throws IOException { ChainSpecificTimeWindowSummary result = null; - Table table = connection.getTable(TableName.valueOf(Config.HBase.TABLE_CHAIN_INFO)); + Table table = connection.getTable(TableName.valueOf(Config.HBase.TABLE_CHAIN_SUMMARY)); Get g = new Get(Bytes.toBytes(key)); Result r = table.get(g); + if (r.rawCells().length == 0) { + return null; + } + result = new ChainSpecificTimeWindowSummary(); for (Cell cell : r.rawCells()) { - result = new ChainSpecificTimeWindowSummary(); if (cell.getValueArray().length > 0) result.addNodeSummaryResult(new ChainNodeSpecificTimeWindowSummary(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()))); @@ -148,4 +155,18 @@ public class HBaseUtil { } } } + + public static void saveChainDetails(List puts) throws IOException, InterruptedException { + Table table = connection.getTable(TableName.valueOf(Config.HBase.TABLE_CHAIN_DETAIL)); + if (puts != null && puts.size() > 0) { + Object[] resultArrays = new Object[puts.size()]; + table.batch(puts, resultArrays); + for (Object result : resultArrays) { + if (result == null) { + //TODO + + } + } + } + } } diff --git a/skywalking-analysis/src/main/resources/config.properties b/skywalking-analysis/src/main/resources/config.properties index a24df42c4..dcf06da68 100644 --- a/skywalking-analysis/src/main/resources/config.properties +++ b/skywalking-analysis/src/main/resources/config.properties @@ -8,7 +8,7 @@ hbase.trace_info_column_family=trace_info traceinfo.trace_info_column_cid=cid -mysql.url= -mysql.username= -mysql.password= -mysql.driverclass= \ No newline at end of file +mysql.url=jdbc:mysql://10.1.228.202:31316/test +mysql.username=devrdbusr21 +mysql.password=devrdbusr21 +mysql.driverclass=com.mysql.jdbc.Driver \ No newline at end of file diff --git a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java index f4e3a6929..78679cc25 100644 --- a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java +++ b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java @@ -4,6 +4,7 @@ package com.ai.cloud.skywalking.analysis.mapper; import com.ai.cloud.skywalking.analysis.config.Config; import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; import com.ai.cloud.skywalking.analysis.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.reduce.ChainInfoReduce; import com.ai.cloud.skywalking.protocol.Span; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; @@ -25,7 +26,8 @@ public class CallChainMapperTest { private static String ZK_QUORUM = "10.1.235.197,10.1.235.198,10.1.235.199"; private static String ZK_CLIENT_PORT = "29181"; - private static String chain_Id = "1.0a2.1452852040127.0664234.11036.55.1"; +// private static String chain_Id = "1.0a2.1453430186581.3efa259.4296.56.1"; + private static String chain_Id = "1.0a2.1453429608422.2701d43.6468.56.1"; private static Configuration configuration = null; private static Connection connection; @@ -35,6 +37,11 @@ public class CallChainMapperTest { ConfigInitializer.initialize(); List spanList = selectByTraceId(chain_Id); ChainInfo chainInfo = CallChainMapper.spanToChainInfo(chain_Id, spanList); + + List chainInfos = new ArrayList(); + chainInfos.add(chainInfo); + + ChainInfoReduce.reduceAction(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken(), chainInfos.iterator()); } public static List selectByTraceId(String traceId) throws IOException {