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

完成mapper部分

This commit is contained in:
zhangxin10 2016-01-18 10:36:40 +08:00
parent ceaf154055
commit d65e8b9a8c
28 changed files with 811 additions and 310 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
skywalking-analysis/pom.xml
View File

@ -41,6 +41,12 @@
<artifactId>skywalking-protocol</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>

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

@ -41,7 +41,7 @@ public class AnalysisServerDriver extends Configured implements Tool {
job.setJarByClass(AnalysisServerDriver.class);
Scan scan = buildHBaseScan(args);
TableMapReduceUtil.initTableMapperJob(Config.HBase.TABLE_NAME, scan, CallChainMapper.class,
TableMapReduceUtil.initTableMapperJob(Config.HBase.CALL_CHAIN_TABLE_NAME, scan, CallChainMapper.class,
String.class, ChainInfo.class, job);
//TableMapReduceUtil.initTableReducerJob("sw-call-chain-model", CallChainReducer.class, job);
return job.waitForCompletion(true) ? 0 : 1;

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

@ -2,6 +2,38 @@ package com.ai.cloud.skywalking.analysis.config;
public class Config {
public static class HBase {
public static String TABLE_NAME;
public static String TRACE_INFO_COLUMN_FAMILY = "trace_info";
public static String CALL_CHAIN_TABLE_NAME;
public static String ZK_QUORUM;
public static String ZK_CLIENT_PORT;
public static String TRACE_INFO_TABLE_NAME;
}
public static class TraceInfo {
public static String PARENT_LEVEL_ID = "parentLevelId";
public static String LEVEL_ID = "levelId";
public static String BUSINESS_KEY = "businessKey";
public static String COST = "cost";
public static String TRACE_INFO_COLUMN_CID = "cid";
public static String STATUS = "status";
public static String USER_ID = "UId";
}
public static class Filter {
public static String FILTER_PACKAGE_NAME;
}
}

86
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ViewPointFilterFactory.java
View File

@ -1,86 +0,0 @@
package com.ai.cloud.skywalking.analysis.config;
import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
public class ViewPointFilterFactory {
private static Logger logger = LoggerFactory.getLogger(ViewPointFilterFactory.class.getName());
private static Map<String, ViewPointFilter> filterType;
private static Object lock = new Object();
private ViewPointFilterFactory() {
//Non
}
private static void initFilterChain() {
Properties properties = new Properties();
try {
properties.load(ViewPointFilterFactory.class.getResourceAsStream("/viewpointfilter.conf"));
} catch (IOException e) {
logger.error("Failed to found the conf file[viewpointfilter.conf]", e);
System.exit(-1);
}
Set<Map.Entry<Object, Object>> entries = properties.entrySet();
for (Map.Entry<Object, Object> entry : entries) {
String types = (String) entry.getKey();
String filters = (String) entry.getValue();
String[] filterClasses = filters.split(">");
ViewPointFilter filter = null;
for (int i = filterClasses.length - 1; i >= 0; i--) {
try {
Class filterClass = Class.forName(filterClasses[i]);
ViewPointFilter newFilter = (ViewPointFilter) filterClass.newInstance();
newFilter.setViewPointFilter(filter);
filter = newFilter;
} catch (ClassNotFoundException e) {
logger.error("Failed to found class[" + filterClasses[i] + "].", e);
System.exit(-1);
} catch (InstantiationException e) {
logger.error("Failed to instance class[" + filterClasses[i] + "].", e);
System.exit(-1);
} catch (IllegalAccessException e) {
logger.error("Failed to access class[" + filterClasses[i] + "].", e);
System.exit(-1);
}
}
String[] type = types.split(",");
for (int i = 0; i < type.length; i++) {
filterType.put(type[i], filter);
}
}
}
public static ViewPointFilter getFilter(String type) {
if (filterType == null) {
synchronized (lock) {
if (filterType == null) {
filterType = new HashMap<String, ViewPointFilter>();
initFilterChain();
}
}
}
ViewPointFilter filter = filterType.get(type);
if (filter == null) {
throw new RuntimeException("Failed to found the filter[" + type + "]");
}
return filter;
}
}

9
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/ChainNodeFilter.java
View File

@ -1,9 +0,0 @@
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.protocol.Span;
public interface ChainNodeFilter {
void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain);
}

49
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/NodeChain.java
View File

@ -1,49 +0,0 @@
package com.ai.cloud.skywalking.analysis.filter;
import com.ai.cloud.skywalking.analysis.filter.impl.CopyAttrNodeFilter;
import com.ai.cloud.skywalking.analysis.filter.impl.CostNodeFilter;
import com.ai.cloud.skywalking.analysis.filter.impl.ViewPointNodeFilter;
import com.ai.cloud.skywalking.analysis.model.ChainNode;
import com.ai.cloud.skywalking.analysis.model.CostMap;
import com.ai.cloud.skywalking.protocol.Span;
import java.util.ArrayList;
import java.util.List;
public class NodeChain {
private int stage = 0;
private static List<ChainNodeFilter> filters = null;
public void doChain(Span span, ChainNode chainNode, CostMap costMap) {
if (filters == null) {
filters = new ArrayList<ChainNodeFilter>();
init();
}
int subscript = stage;
stage = stage + 1;
if (subscript < filters.size()) {
filters.get(subscript).doFilter(span, chainNode, costMap, this);
}
}
private static void init() {
filters.add(new CostNodeFilter());
filters.add(new CopyAttrNodeFilter());
filters.add(new ViewPointNodeFilter());
}
public static void main(String[] args) {
Span span = new Span("1.0a2.1452649597690.035d27f.1608.56.1@~ @~0@~" +
"http://localhost:8080/skywalking-web/order/save@~1452649597692@~5202@~" +
"astraea-PC/192.168.1.102@~0@~ @~W@~true@~ @~1608@~web-application@~6@~A ");
NodeChain nodeChain = new NodeChain();
CostMap costMap = new CostMap();
ChainNode chainNode = new ChainNode();
nodeChain.doChain(span, chainNode, costMap);
System.out.print("xx");
}
}

87
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessChain.java Normal file
View File

@ -0,0 +1,87 @@
package com.ai.cloud.skywalking.analysis.filter;
import com.ai.cloud.skywalking.analysis.config.Config;
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;
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.protocol.Span;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
public class SpanNodeProcessChain {
private static Logger logger = LoggerFactory.getLogger(SpanNodeProcessChain.class.getName());
private static Map<String, SpanNodeProcessFilter> filterMap;
private static Object lock = new Object();
private SpanNodeProcessChain() {
//Non
}
private static void initFilterMap(Map<String, SpanNodeProcessFilter> filterMap) {
Properties properties = new Properties();
try {
properties.load(SpanNodeProcessChain.class.getResourceAsStream("/viewpointfilter.conf"));
} catch (IOException e) {
logger.error("Failed to find config file[viewpointfilter.conf]", e);
System.exit(-1);
}
for (Map.Entry<Object, Object> entry : properties.entrySet()) {
String[] filters = ((String) entry.getValue()).split("->");
String[] types = ((String) entry.getKey()).split(",");
SpanNodeProcessFilter currentFilter = null;
for (int i = filters.length - 1; i >= 0; i--) {
try {
Class filterClass = Class.forName(Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i]);
SpanNodeProcessFilter tmpSpanNodeFilter = (SpanNodeProcessFilter) filterClass.newInstance();
tmpSpanNodeFilter.setNextProcessChain(currentFilter);
currentFilter = tmpSpanNodeFilter;
} catch (ClassNotFoundException e) {
logger.error("Filed to find class[" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "]", e);
System.exit(-1);
} catch (InstantiationException e) {
logger.error("Can not instance class[" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "]", e);
System.exit(-1);
} catch (IllegalAccessException e) {
logger.error("Can not access class[" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "]", e);
System.exit(-1);
} catch (ClassCastException e) {
logger.error("Class [" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "] is not subclass of " +
"com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter", e);
System.exit(-1);
}
}
for (String type : types) {
filterMap.put(type, currentFilter);
}
}
}
public static SpanNodeProcessFilter getProcessChainByCallType(String callType) {
if (filterMap == null) {
synchronized (lock) {
if (filterMap == null) {
filterMap = new HashMap<String, SpanNodeProcessFilter>();
initFilterMap(filterMap);
}
}
}
if (filterMap.containsKey(callType)) {
return filterMap.get(callType);
}
return filterMap.get("default");
}
}

21
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java Normal file
View File

@ -0,0 +1,21 @@
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.protocol.Span;
public abstract class SpanNodeProcessFilter {
private SpanNodeProcessFilter nextProcessChain;
public abstract void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap);
public SpanNodeProcessFilter getNextProcessChain() {
return nextProcessChain;
}
public void setNextProcessChain(SpanNodeProcessFilter nextProcessChain) {
this.nextProcessChain = nextProcessChain;
}
}

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

@ -0,0 +1,18 @@
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;
public class AppendBusinessKeyFilter extends SpanNodeProcessFilter {
@Override
public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) {
node.setViewPoint(node.getViewPoint() + spanEntry.getBusinessKey());
if (getNextProcessChain() != null) {
getNextProcessChain().doFilter(spanEntry, node, costMap);
}
}
}

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

@ -0,0 +1,23 @@
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;
public class CopyAttrFilter extends SpanNodeProcessFilter {
@Override
public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) {
node.setCallType(spanEntry.getCallType().toString());
node.setStatus(spanEntry.getSpanStatus());
node.setLevelId(spanEntry.getLevelId());
node.setParentLevelId(spanEntry.getParentLevelId());
node.setViewPoint(spanEntry.getViewPoint());
node.setUserId(spanEntry.getUserId());
if (getNextProcessChain() != null) {
getNextProcessChain().doFilter(spanEntry, node, costMap);
}
}
}

26
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrNodeFilter.java
View File

@ -1,26 +0,0 @@
package com.ai.cloud.skywalking.analysis.filter.impl;
import com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter;
import com.ai.cloud.skywalking.analysis.filter.NodeChain;
import com.ai.cloud.skywalking.analysis.model.ChainNode;
import com.ai.cloud.skywalking.analysis.model.CostMap;
import com.ai.cloud.skywalking.protocol.Span;
public class CopyAttrNodeFilter implements ChainNodeFilter {
@Override
public void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain) {
node.setCost(span.getCost());
node.setLevelId(span.getLevelId());
node.setParentLevelId(span.getParentLevel());
node.setCallType(span.getCallType());
if (span.getExceptionStack() == null || span.getExceptionStack().length() == 0) {
node.setStatus(ChainNode.NodeStatus.NORMAL);
} else {
node.setStatus(ChainNode.NodeStatus.ABNORMAL);
}
chain.doChain(span, node, costMap);
}
}

34
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CostNodeFilter.java
View File

@ -1,34 +0,0 @@
package com.ai.cloud.skywalking.analysis.filter.impl;
import com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter;
import com.ai.cloud.skywalking.analysis.filter.NodeChain;
import com.ai.cloud.skywalking.analysis.model.ChainNode;
import com.ai.cloud.skywalking.analysis.model.CostMap;
import com.ai.cloud.skywalking.protocol.Span;
public class CostNodeFilter implements ChainNodeFilter {
@Override
public void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain) {
if (span.isReceiver()) {
costMap.put(span.getParentLevel() + "." + span.getLevelId() + "-S", span.getCost());
if (isFirstNode(span)) {
chain.doChain(span, node, costMap);
}
} else {
if (costMap.exists(span.getParentLevel())) {
costMap.put(span.getParentLevel(), costMap.get(span.getParentLevel()) + span.getCost());
} else {
costMap.put(span.getParentLevel(), span.getCost());
}
chain.doChain(span, node, costMap);
}
}
private boolean isFirstNode(Span span) {
return span.getParentLevel().length() == 0 && span.getLevelId() == 0;
}
}

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

@ -0,0 +1,24 @@
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;
public class ProcessCostTimeFilter extends SpanNodeProcessFilter {
@Override
public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) {
long subNodeCost = spanEntry.getCost();
node.setCost(spanEntry.getCost());
if (costMap.exists(spanEntry.getParentLevelId())) {
subNodeCost += costMap.get(spanEntry.getParentLevelId());
}
costMap.put(spanEntry.getParentLevelId(), subNodeCost);
if (getNextProcessChain() != null) {
getNextProcessChain().doFilter(spanEntry, node, costMap);
}
}
}

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

@ -0,0 +1,23 @@
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;
public class ReplaceAddressFilter extends SpanNodeProcessFilter {
//ip:port regex
private static String IP_PORT_REGEX = "(([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))|localhost):([1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5]|[0-9]{1,4})";
@Override
public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) {
String viewPoint = spanEntry.getViewPoint().replaceAll(IP_PORT_REGEX, spanEntry.getApplicationId());
node.setViewPoint(viewPoint);
if (getNextProcessChain() != null) {
getNextProcessChain().doFilter(spanEntry, node, costMap);
}
}
}

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

@ -0,0 +1,23 @@
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.util.TokenGenerator;
public class TokenGenerateFilter extends SpanNodeProcessFilter {
@Override
public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) {
String nodeToken = TokenGenerator.generate(node.getParentLevelId() + "." + node.getLevelId() +
"-" + node.getViewPoint());
node.setNodeToken(nodeToken);
if (getNextProcessChain() != null) {
getNextProcessChain().doFilter(spanEntry, node, costMap);
}
}
}

20
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ViewPointNodeFilter.java
View File

@ -1,20 +0,0 @@
package com.ai.cloud.skywalking.analysis.filter.impl;
import com.ai.cloud.skywalking.analysis.config.ViewPointFilterFactory;
import com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter;
import com.ai.cloud.skywalking.analysis.filter.NodeChain;
import com.ai.cloud.skywalking.analysis.model.ChainNode;
import com.ai.cloud.skywalking.analysis.model.CostMap;
import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter;
import com.ai.cloud.skywalking.protocol.Span;
public class ViewPointNodeFilter implements ChainNodeFilter {
@Override
public void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain) {
ViewPointFilter viewPointFilter = ViewPointFilterFactory.getFilter(span.getSpanType());
String viewPoint = span.getViewPointId();
viewPointFilter.doFilter(span, viewPoint);
node.setViewPoint(viewPoint);
chain.doChain(span, node, costMap);
}
}

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

@ -1,10 +1,13 @@
package com.ai.cloud.skywalking.analysis.mapper;
import com.ai.cloud.skywalking.analysis.filter.NodeChain;
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.protocol.CallType;
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 org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
@ -16,6 +19,10 @@ 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;
public class CallChainMapper extends TableMapper<Text, ChainInfo> {
private Logger logger = LoggerFactory.getLogger(CallChainMapper.class.getName());
@ -23,38 +30,93 @@ public class CallChainMapper extends TableMapper<Text, ChainInfo> {
@Override
protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException,
InterruptedException {
CostMap costMap = new CostMap();
ChainInfo chainInfo = new ChainInfo();
ChainInfo chainInfo = new ChainInfo();
List<Span> spanList = new ArrayList<Span>();
CostMap costMap = new CostMap();
for (Cell cell : value.rawCells()) {
ChainNode node = new ChainNode();
Span span = new Span(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
new NodeChain().doChain(span, node, costMap);
chainInfo.getNodes().add(node);
spanList.add(span);
}
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);
}
//
String firstNodeToken = null;
boolean status = true;
StringBuilder stringBuilder = new StringBuilder();
for (ChainNode node : chainInfo.getNodes()) {
CallType callType = CallType.valueOf(node.getCallType());
switch (callType) {
case ASYNC:
if ((node.getParentLevelId() == null || node.getParentLevelId().length() == 0)
&& node.getLevelId() == 0) {
firstNodeToken = node.getNodeToken();
chainInfo.setUserId(node.getUserId());
}
break;
case SYNC:
// 状态轮询
if (node.getStatus() == ChainNode.NodeStatus.ABNORMAL) {
status = false;
}
break;
case LOCAL:
// 设置时间
computeChainNodeCost(costMap, node);
break;
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
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));
}
}
private Map<String, SpanEntry> mergeSpanDataset(List<Span> spanList) {
Map<String, SpanEntry> spanEntryMap = new HashMap<String, SpanEntry>();
for (Span span : spanList) {
SpanEntry spanEntry = spanEntryMap.get(span.getParentLevel() + "." + span.getLevelId());
if (spanEntry != null && span.isReceiver()) {
if (span.isReceiver()) {
spanEntry.setServerSpan(span);
} else {
spanEntry.setClientSpan(span);
}
} else {
spanEntry = new SpanEntry();
if (span.isReceiver()) {
spanEntry.setServerSpan(span);
} else {
spanEntry.setClientSpan(span);
}
spanEntryMap.put(span.getParentLevel() + "." + span.getLevelId(), spanEntry);
}
}
//入HBase库
//找到首个Node的MD5
//拼接MR任务的Key
context.write(new Text(""), chainInfo);
}
private String convertViewPoint(Span span) {
return null;
return spanEntryMap;
}
}

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

@ -12,6 +12,7 @@ public class ChainInfo implements Writable {
private String chainToken;
private ChainStatus chainStatus;
private List<ChainNode> nodes;
private String userId;
public ChainInfo() {
this.nodes = new ArrayList<ChainNode>();
@ -77,6 +78,14 @@ public class ChainInfo implements Writable {
this.nodes = nodes;
}
public String getUserId() {
return userId;
}
public void setUserId(String userId) {
this.userId = userId;
}
public enum ChainStatus {
NORMAL('N'), ABNORMAL('A');
private char value;

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

@ -5,15 +5,16 @@ public class ChainNode {
private String nodeToken;
private String viewPoint;
//TODO
private String businessKey;
private long cost;
private NodeStatus status;
private String parentLevelId;
private int levelId;
private String callType;
// 不参与序列化
private String userId;
public String getNodeToken() {
return nodeToken;
}
@ -70,6 +71,22 @@ public class ChainNode {
return callType;
}
public void setUserId(String userId) {
this.userId = userId;
}
public String getUserId() {
return userId;
}
public String getBusinessKey() {
return businessKey;
}
public void setBusinessKey(String businessKey) {
this.businessKey = businessKey;
}
public enum NodeStatus {
NORMAL('N'), ABNORMAL('A');
private char value;

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

@ -14,7 +14,7 @@ public class CostMap {
return costs.containsKey(parentLevel);
}
public long get(String parentLevel) {
public Long get(String parentLevel) {
return costs.get(parentLevel);
}
}

157
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java Normal file
View File

@ -0,0 +1,157 @@
package com.ai.cloud.skywalking.analysis.model;
import com.ai.cloud.skywalking.protocol.CallType;
import com.ai.cloud.skywalking.protocol.Span;
public class SpanEntry {
private Span clientSpan;
private Span serverSpan;
public SpanEntry() {
}
public int getLevelId() {
if (clientSpan != null) {
return clientSpan.getLevelId();
}
return serverSpan.getLevelId();
}
public String getParentLevelId() {
if (clientSpan != null) {
return clientSpan.getParentLevel();
}
return serverSpan.getParentLevel();
}
public String getViewPoint() {
if (clientSpan != null) {
return clientSpan.getViewPointId();
}
return serverSpan.getViewPointId();
}
public CallType getCallType() {
if (clientSpan != null) {
return CallType.convert(clientSpan.getCallType());
}
return CallType.convert(serverSpan.getCallType());
}
public long getCost() {
long resultCost = 0;
switch (getCallType()) {
case ASYNC:
resultCost = getClientCost() + getServerCost();
break;
case SYNC:
resultCost = getClientCost();
if (getClientSpan() == null) {
resultCost = getServerCost();
}
break;
case LOCAL:
resultCost = getClientCost();
break;
}
return resultCost;
}
public long getClientCost() {
if (clientSpan != null) {
return clientSpan.getCost();
}
return 0;
}
public long getServerCost() {
if (serverSpan != null) {
return serverSpan.getCost();
}
return 0;
}
public String getBusinessKey() {
if (clientSpan != null) {
return clientSpan.getBusinessKey();
}
return serverSpan.getBusinessKey();
}
public ChainNode.NodeStatus getSpanStatus() {
if (clientSpan != null) {
if (clientSpan.getExceptionStack() != null && clientSpan.getExceptionStack().length() > 0) {
return ChainNode.NodeStatus.ABNORMAL;
}
}
if (serverSpan != null) {
if (serverSpan.getExceptionStack() != null && serverSpan.getExceptionStack().length() > 0) {
return ChainNode.NodeStatus.ABNORMAL;
}
}
return ChainNode.NodeStatus.NORMAL;
}
public String getExceptionStack() {
StringBuilder exceptionStack = new StringBuilder();
if (clientSpan != null) {
exceptionStack.append("Client" + clientSpan.getExceptionStack());
}
if (serverSpan != null) {
exceptionStack.append("Server cause by :" + serverSpan.getExceptionStack());
}
return exceptionStack.toString();
}
public String getApplicationId() {
if (clientSpan != null) {
return clientSpan.getApplicationId();
}
return serverSpan.getApplicationId();
}
public Span getClientSpan() {
return clientSpan;
}
public Span getServerSpan() {
return serverSpan;
}
public void setClientSpan(Span clientSpan) {
this.clientSpan = clientSpan;
}
public void setServerSpan(Span serverSpan) {
this.serverSpan = serverSpan;
}
public String getSpanType() {
if (clientSpan != null) {
return clientSpan.getSpanType();
}
return serverSpan.getSpanType();
}
public String getUserId() {
if (clientSpan != null) {
return clientSpan.getUserId();
}
return serverSpan.getUserId();
}
}

111
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java Normal file
View File

@ -0,0 +1,111 @@
package com.ai.cloud.skywalking.analysis.util;
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 org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
public class HBaseUtil {
private static Logger logger = LoggerFactory.getLogger(HBaseUtil.class.getName());
private static Configuration configuration = null;
private static Connection connection;
public static boolean saveData(String traceId, ChainInfo chainInfo) {
Table table = null;
try {
table = connection.getTable(TableName.valueOf(Config.HBase.TRACE_INFO_TABLE_NAME));
} catch (IOException e) {
logger.error("Cannot found table[" + Config.HBase.TRACE_INFO_TABLE_NAME + "]", e);
}
Put put = new Put(Bytes.toBytes(traceId));
for (ChainNode chainNode : chainInfo.getNodes()) {
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.USER_ID),
Bytes.toBytes(chainNode.getUserId()));
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.STATUS),
Bytes.toBytes(chainNode.getStatus().getValue()));
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.TRACE_INFO_COLUMN_CID),
Bytes.toBytes(chainInfo.getChainToken()));
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.TRACE_INFO_COLUMN_CID),
Bytes.toBytes(chainInfo.getChainToken()));
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.COST),
Bytes.toBytes(chainNode.getCost()));
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.PARENT_LEVEL_ID),
Bytes.toBytes(chainNode.getParentLevelId()));
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.LEVEL_ID),
Bytes.toBytes(chainNode.getLevelId()));
put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY),
Bytes.toBytes(Config.TraceInfo.BUSINESS_KEY),
Bytes.toBytes(chainNode.getBusinessKey()));
}
try {
table.put(put);
if (logger.isDebugEnabled()) {
logger.debug("Insert data[RowKey:{}] success.", put.getId());
}
} catch (IOException e) {
logger.error("Insert data [Rowkey:{}] failed.", put.getId(), e);
return false;
}
return true;
}
static {
try {
initHBaseClient();
Admin admin = connection.getAdmin();
if (!admin.isTableAvailable(TableName.valueOf(Config.HBase.TRACE_INFO_TABLE_NAME))) {
HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(Config.HBase.TRACE_INFO_TABLE_NAME));
tableDesc.addFamily(new HColumnDescriptor(Config.HBase.TRACE_INFO_COLUMN_FAMILY));
admin.createTable(tableDesc);
logger.info("Create table [{}] ok!", Config.HBase.TRACE_INFO_TABLE_NAME);
}
} catch (IOException e) {
logger.error("Create table[{}] failed", Config.HBase.TRACE_INFO_TABLE_NAME, e);
}
}
private static void initHBaseClient() throws IOException {
if (configuration == null) {
configuration = HBaseConfiguration.create();
if (Config.HBase.ZK_QUORUM == null || "".equals(Config.HBase.ZK_QUORUM)) {
logger.error("Miss HBase ZK quorum Configuration", new IllegalArgumentException("Miss HBase ZK quorum Configuration"));
System.exit(-1);
}
configuration.set("hbase.zookeeper.quorum", Config.HBase.ZK_QUORUM);
configuration.set("hbase.zookeeper.property.clientPort", Config.HBase.ZK_CLIENT_PORT);
connection = ConnectionFactory.createConnection(configuration);
}
}
}

18
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/ViewPointFilter.java
View File

@ -1,18 +0,0 @@
package com.ai.cloud.skywalking.analysis.viewpoint;
import com.ai.cloud.skywalking.protocol.Span;
public abstract class ViewPointFilter {
private ViewPointFilter viewPointFilter;
public abstract void doFilter(Span viewPoint, String span);
public ViewPointFilter getViewPointFilter() {
return viewPointFilter;
}
public void setViewPointFilter(ViewPointFilter viewPointFilter) {
this.viewPointFilter = viewPointFilter;
}
}

15
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/AppendBusinessKeyFilter.java
View File

@ -1,15 +0,0 @@
package com.ai.cloud.skywalking.analysis.viewpoint.impl;
import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter;
import com.ai.cloud.skywalking.protocol.Span;
public class AppendBusinessKeyFilter extends ViewPointFilter {
@Override
public void doFilter(Span span, String viewPoint) {
viewPoint += span.getBusinessKey();
if (getViewPointFilter() != null) {
getViewPointFilter().doFilter(span, viewPoint);
}
}
}

22
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/ReplaceIPPortFilter.java
View File

@ -1,22 +0,0 @@
package com.ai.cloud.skywalking.analysis.viewpoint.impl;
import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter;
import com.ai.cloud.skywalking.protocol.Span;
public class ReplaceIPPortFilter extends ViewPointFilter {
//ip:port regex
private static String IP_PORT_REGEX = "^([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\." +
"([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])" +
"|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5])):" +
"([0-9]{1,4}|[1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5])$";
@Override
public void doFilter(Span span, String viewPoint) {
viewPoint = viewPoint.replaceAll(IP_PORT_REGEX, span.getApplicationId());
if (getViewPointFilter() != null) {
getViewPointFilter().doFilter(span, viewPoint);
}
}
}

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

@ -0,0 +1,15 @@
#拦截器的类中的包名
filter.filter_package_name=com.ai.cloud.skywalking.analysis.filter.impl
hbase.call_chain_table_name=sw-call-chain
hbase.zk_quorum=10.1.235.197,10.1.235.198,10.1.235.199
hbase.zk_client_port=29181
hbase.trace_info_table_name=sw-trace-info
hbase.trace_info_column_family=trace_info
traceinfo.parent_level_id=parentLevelId
traceinfo.level_id=levelId
traceinfo.business_key=businessKey
traceinfo.cost=cost
traceinfo.trace_info_column_cid=cid
traceinfo.status=status
traceinfo.user_id=UId

4
skywalking-analysis/src/main/resources/viewpointfilter.conf
View File

@ -1,3 +1,3 @@
#
D,M,W=com.ai.cloud.skywalking.analysis.viewpoint.impl.ReplaceIPPortFilter
J=com.ai.cloud.skywalking.analysis.viewpoint.impl.AppendBusinessKeyFilter
default=CopyAttrFilter->ReplaceAddressFilter->ProcessCostTimeFilter->TokenGenerateFilter
J=CopyAttrFilter->AppendBusinessKeyFilter->ProcessCostTimeFilter->TokenGenerateFilter

152
skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java Normal file
View File

@ -0,0 +1,152 @@
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.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 org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Created by astraea on 2016/1/15.
*/
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 Configuration configuration = null;
private static Connection connection;
@Test
public void testMap() throws Exception {
ConfigInitializer.initialize();
List<Span> spanList = selectByTraceId(chain_Id);
ChainInfo chainInfo = new ChainInfo();
CostMap costMap = new CostMap();
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);
}
//
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
HBaseUtil.saveData(chain_Id, chainInfo);
System.out.println(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));
}
}
private Map<String, SpanEntry> mergeSpanDataset(List<Span> spanList) {
Map<String, SpanEntry> spanEntryMap = new HashMap<String, SpanEntry>();
for (Span span : spanList) {
SpanEntry spanEntry = spanEntryMap.get(span.getParentLevel() + "." + span.getLevelId());
if (spanEntry != null && span.isReceiver()) {
if (span.isReceiver()) {
spanEntry.setServerSpan(span);
} else {
spanEntry.setClientSpan(span);
}
} else {
spanEntry = new SpanEntry();
if (span.isReceiver()) {
spanEntry.setServerSpan(span);
} else {
spanEntry.setClientSpan(span);
}
spanEntryMap.put(span.getParentLevel() + "." + span.getLevelId(), spanEntry);
}
}
return spanEntryMap;
}
public static List<Span> selectByTraceId(String traceId) throws IOException {
List<Span> entries = new ArrayList<Span>();
Table table = connection.getTable(TableName.valueOf(Config.HBase.CALL_CHAIN_TABLE_NAME));
Get g = new Get(Bytes.toBytes(traceId));
Result r = table.get(g);
for (Cell cell : r.rawCells()) {
if (cell.getValueArray().length > 0)
entries.add(new Span(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())));
}
return entries;
}
@Before
public void initHBaseClient() throws IOException {
if (configuration == null) {
configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.quorum", ZK_QUORUM);
configuration.set("hbase.zookeeper.property.clientPort", ZK_CLIENT_PORT);
connection = ConnectionFactory.createConnection(configuration);
}
}
}