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

1.优化代码结构。

This commit is contained in:
wusheng 2016-01-18 11:57:43 +08:00
parent 39383c4ad8
commit 773686a4a0
3 changed files with 69 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -16,7 +16,21 @@ public class ProcessCostTimeFilter extends SpanNodeProcessFilter {
}
costMap.put(spanEntry.getParentLevelId(), subNodeCost);
this.computeChainNodeCost(costMap, node);
this.doNext(spanEntry, node, costMap);
}
private void computeChainNodeCost(CostMap costMap, ChainNode node) {
String levelId = node.getParentLevelId();
if (levelId != null && levelId.length() > 0) {
levelId += ".";
}
levelId += node.getLevelId() + "";
if (costMap.exists(levelId)) {
node.setCost(node.getCost() - costMap.get(levelId));
}
}
}

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

@ -1,14 +1,11 @@
package com.ai.cloud.skywalking.analysis.mapper;
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.analysis.util.TokenGenerator;
import com.ai.cloud.skywalking.protocol.Span;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@ -18,11 +15,14 @@ import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
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;
public class CallChainMapper extends TableMapper<Text, ChainInfo> {
private Logger logger = LoggerFactory.getLogger(CallChainMapper.class.getName());
@ -39,62 +39,21 @@ public class CallChainMapper extends TableMapper<Text, ChainInfo> {
spanList.add(span);
}
//TODO: sort spanList
Map<String, SpanEntry> spanEntryMap = mergeSpanDataset(spanList);
for (Map.Entry<String, SpanEntry> entry : spanEntryMap.entrySet()) {
ChainNode chainNode = new ChainNode();
SpanNodeProcessFilter filter = SpanNodeProcessChain.getProcessChainByCallType(entry.getValue().getSpanType());
filter.doFilter(entry.getValue(), chainNode, costMap);
chainInfo.getNodes().add(chainNode);
chainInfo.addNodes(chainNode);
}
//
String firstNodeToken = null;
boolean status = true;
StringBuilder stringBuilder = new StringBuilder();
for (ChainNode node : chainInfo.getNodes()) {
if ((node.getParentLevelId() == null || node.getParentLevelId().length() == 0)
&& node.getLevelId() == 0) {
firstNodeToken = node.getNodeToken();
chainInfo.setUserId(node.getUserId());
}
// 状态轮询
if (node.getStatus() == ChainNode.NodeStatus.ABNORMAL) {
status = false;
}
// 设置时间
computeChainNodeCost(costMap, node);
stringBuilder.append(node.getParentLevelId() + "." + node.getLevelId() + "-" + node.getNodeToken() + ";");
}
// 设置状态
if (status) {
chainInfo.setChainStatus(ChainInfo.ChainStatus.NORMAL);
} else {
chainInfo.setChainStatus(ChainInfo.ChainStatus.ABNORMAL);
}
// 设置Token
chainInfo.setChainToken(TokenGenerator.generate(stringBuilder.toString()));
//SaveToHbase
chainInfo.generateChainToken();
HBaseUtil.saveData(key.toString(), chainInfo);
context.write(new Text(key.toString() + ":" + firstNodeToken), chainInfo);
}
private void computeChainNodeCost(CostMap costMap, ChainNode node) {
String levelId = node.getParentLevelId();
if (levelId != null && levelId.length() > 0) {
levelId += ".";
}
levelId += node.getLevelId() + "";
if (costMap.get(levelId) != null) {
node.setCost(node.getCost() - costMap.get(levelId));
}
context.write(new Text(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()), chainInfo);
}
private Map<String, SpanEntry> mergeSpanDataset(List<Span> spanList) {

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

@ -2,6 +2,8 @@ package com.ai.cloud.skywalking.analysis.model;
import org.apache.hadoop.io.Writable;
import com.ai.cloud.skywalking.analysis.util.TokenGenerator;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
@ -10,9 +12,10 @@ import java.util.List;
public class ChainInfo implements Writable {
private String chainToken;
private ChainStatus chainStatus;
private ChainStatus chainStatus = ChainStatus.NORMAL;
private List<ChainNode> nodes;
private String userId;
private String userId = null;
private ChainNode firstChainNode;
public ChainInfo() {
this.nodes = new ArrayList<ChainNode>();
@ -53,13 +56,24 @@ public class ChainInfo implements Writable {
}
}
public String getChainToken() {
return chainToken;
}
public String getEntranceNodeToken(){
if(firstChainNode == null){
return "";
}else{
return firstChainNode.getNodeToken();
}
}
public void setChainToken(String chainToken) {
this.chainToken = chainToken;
public void generateChainToken() {
StringBuilder chainTokenDesc = new StringBuilder();
for(ChainNode node: nodes){
chainTokenDesc.append(node.getParentLevelId() + "." + node.getLevelId() + "-" + node.getNodeToken() + ";");
}
this.chainToken = TokenGenerator.generate(chainTokenDesc.toString());
}
public ChainStatus getChainStatus() {
@ -69,15 +83,25 @@ public class ChainInfo implements Writable {
public void setChainStatus(ChainStatus chainStatus) {
this.chainStatus = chainStatus;
}
public void addNodes(ChainNode chainNode){
this.nodes.add(0, chainNode);
if (chainNode.getStatus() == ChainNode.NodeStatus.ABNORMAL) {
chainStatus = ChainStatus.ABNORMAL;
}
if(userId == null){
userId = chainNode.getUserId();
}
if ((chainNode.getParentLevelId() == null || chainNode.getParentLevelId().length() == 0)
&& chainNode.getLevelId() == 0) {
firstChainNode = chainNode;
}
}
public List<ChainNode> getNodes() {
return nodes;
}
public void setNodes(List<ChainNode> nodes) {
this.nodes = nodes;
}
public String getUserId() {
return userId;
}
@ -85,7 +109,7 @@ public class ChainInfo implements Writable {
public void setUserId(String userId) {
this.userId = userId;
}
public enum ChainStatus {
NORMAL('N'), ABNORMAL('A');
private char value;