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

完成Reduce功能

This commit is contained in:
ascrutae 2016-01-22 15:54:25 +08:00
parent 7ec811dffc
commit cffcb51be2
28 changed files with 282 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
skywalking-analysis/pom.xml
View File

@ -52,6 +52,11 @@
<artifactId>gson</artifactId>
<version>2.2.2</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.38</version>
</dependency>
</dependencies>
<build>

12
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java
View File

@ -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 {

44
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/dao/CallChainInfoDao.java
View File

@ -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<String, Timestamp> updateChainInfo) throws SQLException {
PreparedStatement 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) {
//TODO
}
preparedStatement.close();
connection.commit();
}
}

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java
View File

@ -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 {

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java
View File

@ -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 {

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java
View File

@ -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 {

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java
View File

@ -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

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java
View File

@ -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 {

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java
View File

@ -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 {

2
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java
View File

@ -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;

2
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CostMap.java
View File

@ -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;

3
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/SpanEntry.java
View File

@ -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;

12
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainDetail.java
View File

@ -1,12 +0,0 @@
package com.ai.cloud.skywalking.analysis.model;
public class ChainDetail {
public ChainDetail(ChainInfo chainInfo) {
}
public ChainDetail(UncategorizeChainInfo child) {
}
}

10
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java
View File

@ -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;
}
}

20
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java
View File

@ -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);
}
}

9
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CategorizedChainInfo.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/CategorizedChainInfo.java
View File

@ -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"),

53
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainDetail.java Normal file
View File

@ -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<String, ChainNode> chainNodeMap = new HashMap<String, ChainNode>();
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<String, ChainNode> entry : chainNodeMap.entrySet()){
put.addColumn(Config.HBase.TRACE_DETAIL_FAMILY_COLUMN.getBytes(),entry.getKey().getBytes(),
entry.getValue().toString().getBytes());
}
CallChainInfoDao.saveChainDetail(this);
}
public Collection<ChainNode> getChainNodes() {
return chainNodeMap.values();
}
public String getUserId() {
return userId;
}
public String getChainToken() {
return chainToken;
}
}

37
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java
View File

@ -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<Text, ChainInfo, Put> {
@Override
protected void reduce(Text key, Iterable<ChainInfo> values, Context context) throws IOException, InterruptedException {
ChainRelate chainRelate = HBaseUtil.selectCallChainRelationship(key.toString());
Summary summary = new Summary();
Iterator<ChainInfo> 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<ChainInfo> 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表
}
}

25
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNodeSpecificTimeWindowSummary.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainNodeSpecificTimeWindowSummary.java
View File

@ -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<Map<String, SummaryResult>>() {
}.getType());
}
@ -36,23 +38,12 @@ public class ChainNodeSpecificTimeWindowSummary {
return traceLevelId;
}
public void setTraceLevelId(String traceLevelId) {
this.traceLevelId = traceLevelId;
}
public Map<String, SummaryResult> getSummerResultMap() {
return summerResultMap;
}
public void setSummerResultMap(Map<String, SummaryResult> 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);
}

43
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainRelate.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainRelate.java
View File

@ -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<String, CategorizedChainInfo> 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<Put> puts = new ArrayList<Put>();
for (Map.Entry<String, ChainDetail> 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);
}

3
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainSpecificTimeWindowSummary.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainSpecificTimeWindowSummary.java
View File

@ -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;

33
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/Summary.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/Summary.java
View File

@ -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<String, ChainSpecificTimeWindowSummary> summaryResultMap;
private Map<String, Timestamp> updateChainInfo;
public Summary() {
summaryResultMap = new HashMap<String, ChainSpecificTimeWindowSummary>();
updateChainInfo = new HashMap<String, Timestamp>();
}
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<Put> puts = new ArrayList<Put>();
for (Map.Entry<String, ChainSpecificTimeWindowSummary> entry : summaryResultMap.entrySet()) {
Put put = new Put(entry.getKey().getBytes());

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SummaryResult.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/SummaryResult.java
View File

@ -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;

29
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/UncategorizeChainInfo.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/UncategorizeChainInfo.java
View File

@ -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);
}
}

16
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/ChainInfoUtil.java
View File

@ -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;
}
}

29
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java
View File

@ -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<Put> 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
}
}
}
}
}

8
skywalking-analysis/src/main/resources/config.properties
View File

@ -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=
mysql.url=jdbc:mysql://10.1.228.202:31316/test
mysql.username=devrdbusr21
mysql.password=devrdbusr21
mysql.driverclass=com.mysql.jdbc.Driver

9
skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java
View File

@ -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<Span> spanList = selectByTraceId(chain_Id);
ChainInfo chainInfo = CallChainMapper.spanToChainInfo(chain_Id, spanList);
List<ChainInfo> chainInfos = new ArrayList<ChainInfo>();
chainInfos.add(chainInfo);
ChainInfoReduce.reduceAction(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken(), chainInfos.iterator());
}
public static List<Span> selectByTraceId(String traceId) throws IOException {