1.根据全新的分析逻辑，提交Mapper任务的基础结构代码。最终将移除原有的mr任务以及相关代码。
This commit is contained in:
wusheng
parent
ea077348a3
commit
1b60750ea9
|
|
@ -23,7 +23,7 @@ import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainInfo;
|
|||
import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainNode;
|
||||
import com.ai.cloud.skywalking.analysis.categorize2chain.util.HBaseUtil;
|
||||
import com.ai.cloud.skywalking.analysis.categorize2chain.util.SubLevelSpanCostCounter;
|
||||
import com.ai.cloud.skywalking.analysis.categorize2chain.util.VersionIdentifier;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.VersionIdentifier;
|
||||
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;
|
||||
import com.ai.cloud.skywalking.protocol.Span;
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import com.ai.cloud.skywalking.analysis.categorize2chain.SpanEntry;
|
|||
import com.ai.cloud.skywalking.analysis.categorize2chain.filter.SpanNodeProcessFilter;
|
||||
import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainNode;
|
||||
import com.ai.cloud.skywalking.analysis.categorize2chain.util.SubLevelSpanCostCounter;
|
||||
import com.ai.cloud.skywalking.analysis.categorize2chain.util.TokenGenerator;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator;
|
||||
|
||||
public class TokenGenerateFilter extends SpanNodeProcessFilter {
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
package com.ai.cloud.skywalking.analysis.categorize2chain.po;
|
||||
|
||||
import com.ai.cloud.skywalking.analysis.categorize2chain.util.TokenGenerator;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator;
|
||||
import com.google.gson.Gson;
|
||||
import com.google.gson.JsonObject;
|
||||
import com.google.gson.JsonParser;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,54 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.hadoop.hbase.Cell;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
|
||||
import org.apache.hadoop.hbase.mapreduce.TableMapper;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.apache.hadoop.io.Text;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.entity.TraceSpanTree;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.VersionIdentifier;
|
||||
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;
|
||||
import com.ai.cloud.skywalking.protocol.Span;
|
||||
|
||||
public class ChainBuildMapper extends TableMapper<Text, Text> {
|
||||
private Logger logger = LoggerFactory
|
||||
.getLogger(ChainBuildMapper.class);
|
||||
|
||||
@Override
|
||||
protected void setup(Context context) throws IOException,
|
||||
InterruptedException {
|
||||
ConfigInitializer.initialize();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void map(ImmutableBytesWritable key, Result value, Context context)
|
||||
throws IOException, InterruptedException {
|
||||
if(!VersionIdentifier.enableAnaylsis(Bytes.toString(key.get()))){
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
List<Span> spanList = new ArrayList<Span>();
|
||||
for (Cell cell : value.rawCells()) {
|
||||
Span span = new Span(Bytes.toString(cell.getValueArray(),
|
||||
cell.getValueOffset(), cell.getValueLength()));
|
||||
spanList.add(span);
|
||||
|
||||
}
|
||||
|
||||
TraceSpanTree tree = new TraceSpanTree();
|
||||
tree.build(spanList);
|
||||
} catch (Throwable e) {
|
||||
logger.error("Failed to mapper call chain[" + key.toString() + "]",
|
||||
e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,200 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild.entity;
|
||||
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.StringUtil;
|
||||
import com.ai.cloud.skywalking.protocol.CallType;
|
||||
import com.ai.cloud.skywalking.protocol.Span;
|
||||
|
||||
public class TraceSpanNode {
|
||||
protected TraceSpanNode prev = null;
|
||||
|
||||
protected TraceSpanNode next = null;
|
||||
|
||||
protected TraceSpanNode parent = null;
|
||||
|
||||
protected TraceSpanNode sub = null;
|
||||
|
||||
protected boolean visualNode = true;
|
||||
|
||||
protected String parentLevel;
|
||||
|
||||
protected int levelId;
|
||||
|
||||
protected String viewPointId = "";
|
||||
|
||||
protected long cost = 0;
|
||||
|
||||
protected long callTimes = 0;
|
||||
|
||||
/**
|
||||
* 节点调用的状态<br/>
|
||||
* 0：成功<br/>
|
||||
* 1：异常<br/>
|
||||
* 异常判断原则：代码产生exception，并且此exception不在忽略列表中
|
||||
*/
|
||||
protected byte statusCode = 0;
|
||||
|
||||
/**
|
||||
* 节点调用的错误堆栈<br/>
|
||||
* 堆栈以JAVA的exception为主要判断依据
|
||||
*/
|
||||
protected String exceptionStack;
|
||||
/**
|
||||
* 节点类型描述<br/>
|
||||
* 已字符串的形式描述<br/>
|
||||
* 如：java,dubbo等
|
||||
*/
|
||||
protected String spanType = "";
|
||||
|
||||
/**
|
||||
* 节点调用过程中的业务字段<br/>
|
||||
* 如：业务系统设置的订单号，SQL语句等
|
||||
*/
|
||||
protected String businessKey = "";
|
||||
|
||||
/**
|
||||
* 节点调用所在的系统逻辑名称<br/>
|
||||
* 由授权文件指定
|
||||
*/
|
||||
protected String applicationId = "";
|
||||
|
||||
public TraceSpanNode(TraceSpanNode parent, TraceSpanNode sub, TraceSpanNode prev, TraceSpanNode next, Span span) {
|
||||
this(parent, sub, prev, next);
|
||||
this.visualNode = false;
|
||||
this.parentLevel = span.getParentLevel();
|
||||
this.levelId = span.getLevelId();
|
||||
this.viewPointId = span.getViewPointId();
|
||||
this.cost = span.getCost();
|
||||
this.callTimes = 1;
|
||||
this.statusCode = span.getStatusCode();
|
||||
if(span.isReceiver()){
|
||||
this.exceptionStack = "server stack:";
|
||||
}else{
|
||||
this.exceptionStack = "client stack:";
|
||||
}
|
||||
this.exceptionStack += span.getExceptionStack();
|
||||
this.spanType = span.getSpanType();
|
||||
this.businessKey = span.getBusinessKey();
|
||||
this.applicationId = span.getApplicationId();
|
||||
}
|
||||
|
||||
public TraceSpanNode(TraceSpanNode parent, TraceSpanNode sub, TraceSpanNode prev, TraceSpanNode next){
|
||||
this.visualNode = true;
|
||||
this.parent = parent;
|
||||
if(parent != null){
|
||||
parent.sub = this;
|
||||
}
|
||||
this.sub = sub;
|
||||
if(sub != null){
|
||||
sub.parent = this;
|
||||
}
|
||||
this.prev = prev;
|
||||
if(prev != null){
|
||||
prev.next = this;
|
||||
}
|
||||
this.next = next;
|
||||
if(next != null){
|
||||
next.prev = this;
|
||||
}
|
||||
}
|
||||
|
||||
protected TraceSpanNode(TraceSpanNode parent, TraceSpanNode sub, TraceSpanNode prev, TraceSpanNode next, String parentLevelId, int levelId){
|
||||
this(parent, sub, prev, next);
|
||||
this.parentLevel = parentLevelId;
|
||||
this.levelId = levelId;
|
||||
this.callTimes = 0;
|
||||
}
|
||||
|
||||
boolean hasNext(){
|
||||
if(this.next != null){
|
||||
return true;
|
||||
}else{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
boolean hasSub(){
|
||||
if(this.sub != null){
|
||||
return true;
|
||||
}else{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
void mergeSpan(Span span){
|
||||
if(CallType.convert(span.getCallType()) == CallType.ASYNC){
|
||||
this.cost += span.getCost();
|
||||
}
|
||||
if(span.getStatusCode() != 0 && !StringUtil.isBlank(span.getExceptionStack())){
|
||||
if(span.isReceiver()){
|
||||
this.exceptionStack += "server stack:";
|
||||
}else{
|
||||
this.exceptionStack += "client stack:";
|
||||
}
|
||||
this.exceptionStack += span.getExceptionStack();
|
||||
}
|
||||
}
|
||||
|
||||
TraceSpanNode next(){
|
||||
return this.next;
|
||||
}
|
||||
|
||||
TraceSpanNode sub(){
|
||||
return this.sub;
|
||||
}
|
||||
|
||||
public TraceSpanNode getPrev() {
|
||||
return prev;
|
||||
}
|
||||
|
||||
public TraceSpanNode getNext() {
|
||||
return next;
|
||||
}
|
||||
|
||||
public TraceSpanNode getParent() {
|
||||
return parent;
|
||||
}
|
||||
|
||||
public TraceSpanNode getSub() {
|
||||
return sub;
|
||||
}
|
||||
|
||||
public boolean isVisualNode() {
|
||||
return visualNode;
|
||||
}
|
||||
|
||||
public String getParentLevel() {
|
||||
return parentLevel;
|
||||
}
|
||||
|
||||
public int getLevelId() {
|
||||
return levelId;
|
||||
}
|
||||
|
||||
public String getViewPointId() {
|
||||
return viewPointId;
|
||||
}
|
||||
|
||||
public long getCost() {
|
||||
return cost;
|
||||
}
|
||||
|
||||
public byte getStatusCode() {
|
||||
return statusCode;
|
||||
}
|
||||
|
||||
public String getExceptionStack() {
|
||||
return exceptionStack;
|
||||
}
|
||||
|
||||
public String getSpanType() {
|
||||
return spanType;
|
||||
}
|
||||
|
||||
public String getBusinessKey() {
|
||||
return businessKey;
|
||||
}
|
||||
|
||||
public String getApplicationId() {
|
||||
return applicationId;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,134 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild.entity;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.exception.BuildTraceSpanTreeException;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.StringUtil;
|
||||
import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator;
|
||||
import com.ai.cloud.skywalking.protocol.Span;
|
||||
|
||||
public class TraceSpanTree {
|
||||
private Logger logger = LoggerFactory.getLogger(TraceSpanTree.class);
|
||||
|
||||
private String userId = null;
|
||||
|
||||
private String cid;
|
||||
|
||||
private TraceSpanNode treeRoot;
|
||||
|
||||
public TraceSpanTree() {
|
||||
}
|
||||
|
||||
public String build(List<Span> spanList) throws BuildTraceSpanTreeException {
|
||||
if (spanList.size() == 0) {
|
||||
throw new BuildTraceSpanTreeException("spanList is empty.");
|
||||
}
|
||||
|
||||
Collections.sort(spanList, new Comparator<Span>() {
|
||||
@Override
|
||||
public int compare(Span span1, Span span2) {
|
||||
String span1TraceLevel = span1.getParentLevel() + "."
|
||||
+ span1.getLevelId();
|
||||
String span2TraceLevel = span2.getParentLevel() + "."
|
||||
+ span2.getLevelId();
|
||||
return span1TraceLevel.compareTo(span2TraceLevel);
|
||||
}
|
||||
});
|
||||
cid = generateChainToken(spanList.get(0));
|
||||
treeRoot = new TraceSpanNode(null, null, null, null, spanList.get(0));
|
||||
if (spanList.size() > 1) {
|
||||
for (int i = 1; i < spanList.size(); i++) {
|
||||
this.build(spanList.get(i));
|
||||
}
|
||||
}
|
||||
|
||||
return cid;
|
||||
}
|
||||
|
||||
private void build(Span span) throws BuildTraceSpanTreeException {
|
||||
if (userId == null && span.getUserId() != null) {
|
||||
userId = span.getUserId();
|
||||
}
|
||||
|
||||
TraceSpanNode clientOrServerNode = findNodeAndCreateVisualNodeIfNess(
|
||||
span.getParentLevel(), span.getLevelId());
|
||||
if (clientOrServerNode != null) {
|
||||
clientOrServerNode.mergeSpan(span);
|
||||
}
|
||||
|
||||
if (span.getLevelId() > 0) {
|
||||
TraceSpanNode foundNode = findNodeAndCreateVisualNodeIfNess(
|
||||
span.getParentLevel(), span.getLevelId() - 1);
|
||||
if (foundNode != null) {
|
||||
new TraceSpanNode(null, null, foundNode, foundNode.next(), span);
|
||||
}
|
||||
} else {
|
||||
/**
|
||||
* levelId=0 find for parent level if parentLevelId = 0.0.1 then
|
||||
* find node[parentLevelId=0.0,levelId=1]
|
||||
*/
|
||||
String parentLevel = span.getParentLevel();
|
||||
int idx = parentLevel.lastIndexOf("\\.");
|
||||
if (idx < 0) {
|
||||
throw new BuildTraceSpanTreeException("parentLevel="
|
||||
+ parentLevel + " is unexpected.");
|
||||
}
|
||||
TraceSpanNode foundNode = findNodeAndCreateVisualNodeIfNess(
|
||||
parentLevel.substring(0, idx),
|
||||
Integer.parseInt(parentLevel.substring(idx + 1)));
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private TraceSpanNode findNodeAndCreateVisualNodeIfNess(
|
||||
String parentLevelId, int levelId) {
|
||||
String levelDesc = StringUtil.isBlank(parentLevelId) ? (levelId + "")
|
||||
: (parentLevelId + "." + levelId);
|
||||
String[] levelArray = levelDesc.split("\\.");
|
||||
|
||||
TraceSpanNode currentNode = treeRoot;
|
||||
String contextParentLevelId = "";
|
||||
for (String currentLevel : levelArray) {
|
||||
int currentLevelInt = Integer.parseInt(currentLevel);
|
||||
for (int i = 0; i < currentLevelInt; i++) {
|
||||
if (currentNode.hasNext()) {
|
||||
currentNode = currentNode.next();
|
||||
} else {
|
||||
// create visual next node
|
||||
currentNode = new VisualTraceSpanNode(null, null,
|
||||
currentNode, null, contextParentLevelId, i);
|
||||
}
|
||||
}
|
||||
contextParentLevelId = contextParentLevelId == "" ? ("" + currentLevelInt)
|
||||
: (contextParentLevelId + "." + currentLevelInt);
|
||||
if (currentNode.hasSub()) {
|
||||
currentNode = currentNode.sub();
|
||||
} else {
|
||||
// create visual sub node
|
||||
currentNode = new VisualTraceSpanNode(currentNode, null, null,
|
||||
null, contextParentLevelId, 0);
|
||||
}
|
||||
}
|
||||
|
||||
return currentNode;
|
||||
}
|
||||
|
||||
private String generateChainToken(Span level0Span)
|
||||
throws BuildTraceSpanTreeException {
|
||||
if (StringUtil.isBlank(level0Span.getParentLevel())
|
||||
&& level0Span.getLevelId() == 0) {
|
||||
StringBuilder chainTokenDesc = new StringBuilder();
|
||||
chainTokenDesc.append(level0Span.getViewPointId());
|
||||
return TokenGenerator.generateCID(chainTokenDesc.toString());
|
||||
} else {
|
||||
throw new BuildTraceSpanTreeException("tid:"
|
||||
+ level0Span.getTraceId() + " level0 span data is illegal");
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,12 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild.entity;
|
||||
|
||||
public class VisualTraceSpanNode extends TraceSpanNode {
|
||||
|
||||
protected VisualTraceSpanNode(TraceSpanNode parent, TraceSpanNode sub,
|
||||
TraceSpanNode prev, TraceSpanNode next, String parentLevelId,
|
||||
int levelId) {
|
||||
super(parent, sub, prev, next, parentLevelId, levelId);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild.exception;
|
||||
|
||||
public class BuildTraceSpanTreeException extends Exception {
|
||||
private static final long serialVersionUID = 5816399370389190974L;
|
||||
|
||||
public BuildTraceSpanTreeException(String msg){
|
||||
super(msg);
|
||||
}
|
||||
|
||||
public BuildTraceSpanTreeException(String msg, Exception cause){
|
||||
super(msg, cause);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
package com.ai.cloud.skywalking.analysis.chainbuild.util;
|
||||
|
||||
public class StringUtil {
|
||||
public static boolean isBlank(String str){
|
||||
if(str == null || str == "" || str.trim() == ""){
|
||||
return true;
|
||||
}else{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public static boolean equal(String str1, String str2){
|
||||
if(str1 == null){
|
||||
str1 = "";
|
||||
}
|
||||
if(str2 == null){
|
||||
str2 = "";
|
||||
}
|
||||
|
||||
return str1.trim().equals(str2.trim());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package com.ai.cloud.skywalking.analysis.categorize2chain.util;
|
||||
package com.ai.cloud.skywalking.analysis.chainbuild.util;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package com.ai.cloud.skywalking.analysis.categorize2chain.util;
|
||||
package com.ai.cloud.skywalking.analysis.chainbuild.util;
|
||||
|
||||
/**
|
||||
* 版本识别器
|
||||
Loading…
Reference in New Issue