From 1b60750ea94fb76b3f1e3ab135c68ea0c0d5cbc7 Mon Sep 17 00:00:00 2001 From: wusheng Date: Tue, 1 Mar 2016 17:27:02 +0800 Subject: [PATCH] =?UTF-8?q?1.=E6=A0=B9=E6=8D=AE=E5=85=A8=E6=96=B0=E7=9A=84?= =?UTF-8?q?=E5=88=86=E6=9E=90=E9=80=BB=E8=BE=91=EF=BC=8C=E6=8F=90=E4=BA=A4?= =?UTF-8?q?Mapper=E4=BB=BB=E5=8A=A1=E7=9A=84=E5=9F=BA=E7=A1=80=E7=BB=93?= =?UTF-8?q?=E6=9E=84=E4=BB=A3=E7=A0=81=E3=80=82=E6=9C=80=E7=BB=88=E5=B0=86?= =?UTF-8?q?=E7=A7=BB=E9=99=A4=E5=8E=9F=E6=9C=89=E7=9A=84mr=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E4=BB=A5=E5=8F=8A=E7=9B=B8=E5=85=B3=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Categorize2ChainMapper.java | 2 +- .../filter/impl/TokenGenerateFilter.java | 2 +- .../categorize2chain/po/ChainInfo.java | 2 +- .../analysis/chainbuild/ChainBuildMapper.java | 54 +++++ .../chainbuild/entity/TraceSpanNode.java | 200 ++++++++++++++++++ .../chainbuild/entity/TraceSpanTree.java | 134 ++++++++++++ .../entity/VisualTraceSpanNode.java | 12 ++ .../BuildTraceSpanTreeException.java | 13 ++ .../analysis/chainbuild/util/StringUtil.java | 22 ++ .../util/TokenGenerator.java | 2 +- .../util/VersionIdentifier.java | 2 +- 11 files changed, 440 insertions(+), 5 deletions(-) create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildMapper.java create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanNode.java create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanTree.java create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/VisualTraceSpanNode.java create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/BuildTraceSpanTreeException.java create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/StringUtil.java rename skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/{categorize2chain => chainbuild}/util/TokenGenerator.java (95%) rename skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/{categorize2chain => chainbuild}/util/VersionIdentifier.java (89%) diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java index d788dce60..14248280e 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java @@ -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; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/filter/impl/TokenGenerateFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/filter/impl/TokenGenerateFilter.java index ac8c6aa0f..990de051d 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/filter/impl/TokenGenerateFilter.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/filter/impl/TokenGenerateFilter.java @@ -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 { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/po/ChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/po/ChainInfo.java index ae8d0252a..0a0a14294 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/po/ChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/po/ChainInfo.java @@ -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; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildMapper.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildMapper.java new file mode 100644 index 000000000..fbac6e728 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildMapper.java @@ -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 { + 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 spanList = new ArrayList(); + 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); + } + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanNode.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanNode.java new file mode 100644 index 000000000..3110891d5 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanNode.java @@ -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; + + /** + * 节点调用的状态
+ * 0：成功
+ * 1：异常
+ * 异常判断原则：代码产生exception，并且此exception不在忽略列表中 + */ + protected byte statusCode = 0; + + /** + * 节点调用的错误堆栈
+ * 堆栈以JAVA的exception为主要判断依据 + */ + protected String exceptionStack; + /** + * 节点类型描述
+ * 已字符串的形式描述
+ * 如：java,dubbo等 + */ + protected String spanType = ""; + + /** + * 节点调用过程中的业务字段
+ * 如：业务系统设置的订单号，SQL语句等 + */ + protected String businessKey = ""; + + /** + * 节点调用所在的系统逻辑名称
+ * 由授权文件指定 + */ + 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; + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanTree.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanTree.java new file mode 100644 index 000000000..e7a1d7d4b --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/TraceSpanTree.java @@ -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 spanList) throws BuildTraceSpanTreeException { + if (spanList.size() == 0) { + throw new BuildTraceSpanTreeException("spanList is empty."); + } + + Collections.sort(spanList, new Comparator() { + @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"); + } + } + +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/VisualTraceSpanNode.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/VisualTraceSpanNode.java new file mode 100644 index 000000000..d1ce42fec --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/VisualTraceSpanNode.java @@ -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); + } + + +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/BuildTraceSpanTreeException.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/BuildTraceSpanTreeException.java new file mode 100644 index 000000000..ba0d2fcf8 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/BuildTraceSpanTreeException.java @@ -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); + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/StringUtil.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/StringUtil.java new file mode 100644 index 000000000..c24615f29 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/StringUtil.java @@ -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()); + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/util/TokenGenerator.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/TokenGenerator.java similarity index 95% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/util/TokenGenerator.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/TokenGenerator.java index 32185a18a..bde561d82 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/util/TokenGenerator.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/TokenGenerator.java @@ -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; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/util/VersionIdentifier.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/VersionIdentifier.java similarity index 89% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/util/VersionIdentifier.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/VersionIdentifier.java index a24b236fd..791da5269 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/util/VersionIdentifier.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/util/VersionIdentifier.java @@ -1,4 +1,4 @@ -package com.ai.cloud.skywalking.analysis.categorize2chain.util; +package com.ai.cloud.skywalking.analysis.chainbuild.util; /** * 版本识别器