From 889fcb9007935c70a7aa63b635c918017e7f1cb3 Mon Sep 17 00:00:00 2001 From: wusheng Date: Mon, 7 Mar 2016 15:04:40 +0800 Subject: [PATCH] =?UTF-8?q?1.=E4=BF=AE=E6=94=B9=E9=83=A8=E5=88=86=E5=88=86?= =?UTF-8?q?=E6=9E=90=E4=BB=A3=E7=A0=81=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analysis/chainbuild/ChainBuildMapper.java | 179 +++++++++--------- .../chainbuild/ChainBuildReducer.java | 62 +++--- .../exception/Tid2CidECovertException.java | 14 ++ .../analysis/chainbuild/po/ChainInfo.java | 64 +++---- .../analysis/mapper/CallChainMapperTest.java | 37 ++-- 5 files changed, 188 insertions(+), 168 deletions(-) create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/Tid2CidECovertException.java 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 index b950fb270..d862d80ee 100644 --- 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 @@ -1,13 +1,13 @@ package com.ai.cloud.skywalking.analysis.chainbuild; -import com.ai.cloud.skywalking.analysis.chainbuild.filter.SpanNodeProcessChain; -import com.ai.cloud.skywalking.analysis.chainbuild.filter.SpanNodeProcessFilter; -import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo; -import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainNode; -import com.ai.cloud.skywalking.analysis.chainbuild.util.SubLevelSpanCostCounter; -import com.ai.cloud.skywalking.analysis.chainbuild.util.VersionIdentifier; -import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; -import com.ai.cloud.skywalking.protocol.Span; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +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; @@ -17,90 +17,99 @@ import org.apache.hadoop.io.Text; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.util.*; +import com.ai.cloud.skywalking.analysis.chainbuild.exception.Tid2CidECovertException; +import com.ai.cloud.skywalking.analysis.chainbuild.filter.SpanNodeProcessChain; +import com.ai.cloud.skywalking.analysis.chainbuild.filter.SpanNodeProcessFilter; +import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo; +import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainNode; +import com.ai.cloud.skywalking.analysis.chainbuild.util.SubLevelSpanCostCounter; +import com.ai.cloud.skywalking.analysis.chainbuild.util.VersionIdentifier; +import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; +import com.ai.cloud.skywalking.protocol.Span; +import com.google.gson.Gson; -public class ChainBuildMapper extends TableMapper { +public class ChainBuildMapper extends TableMapper { - private Logger logger = LoggerFactory - .getLogger(ChainBuildMapper.class.getName()); + private Logger logger = LoggerFactory.getLogger(ChainBuildMapper.class); - @Override - protected void setup(Context context) throws IOException, - InterruptedException { - ConfigInitializer.initialize(); - } + @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; + } - @Override - protected void map(ImmutableBytesWritable key, Result value, Context context) - throws IOException, InterruptedException { - if (!VersionIdentifier.enableAnaylsis(Bytes.toString(key.get()))) { - return; - } + List spanList = new ArrayList(); + ChainInfo chainInfo = null; + try { + for (Cell cell : value.rawCells()) { + Span span = new Span(Bytes.toString(cell.getValueArray(), + cell.getValueOffset(), cell.getValueLength())); + spanList.add(span); + } + if (spanList.size() == 0) { + throw new Tid2CidECovertException("tid[" + + Bytes.toString(key.get()) + "] has no span data."); + } - List spanList = new ArrayList(); - ChainInfo chainInfo = null; - try { - for (Cell cell : value.rawCells()) { - Span span = new Span(Bytes.toString(cell.getValueArray(), - cell.getValueOffset(), cell.getValueLength())); - spanList.add(span); - } + chainInfo = spanToChainInfo(Bytes.toString(key.get()), spanList); + logger.debug("convert tid[" + Bytes.toString(key.get()) + + "] to chain with cid[" + chainInfo.getCID() + "]."); + context.write( + new Text(chainInfo.getEntranceNodeToken()), new Text(new Gson().toJson(chainInfo))); + } catch (Exception e) { + logger.error("Failed to mapper call chain[" + key.toString() + "]", + e); + } + } - chainInfo = spanToChainInfo(Bytes.toString(key.get()), spanList); - logger.info("Success convert span to chain info...." - + chainInfo.getCID() + " TraceId : " + Bytes.toString(key.get())); - context.write( - new Text(chainInfo.getUserId() + ":" - + chainInfo.getEntranceNodeToken()), chainInfo); - } catch (Exception e) { - logger.error("Failed to mapper call chain[" + key.toString() + "]", - e); - } - } + public static ChainInfo spanToChainInfo(String tid, List spanList) { + SubLevelSpanCostCounter costMap = new SubLevelSpanCostCounter(); + ChainInfo chainInfo = new ChainInfo(tid); + 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); + } + }); - public static ChainInfo spanToChainInfo(String key, List spanList) { - SubLevelSpanCostCounter costMap = new SubLevelSpanCostCounter(); - ChainInfo chainInfo = new ChainInfo(); - 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); - } - }); + Map spanEntryMap = mergeSpanDataSet(spanList); + for (Map.Entry entry : spanEntryMap.entrySet()) { + ChainNode chainNode = new ChainNode(); + SpanNodeProcessFilter filter = SpanNodeProcessChain + .getProcessChainByCallType(entry.getValue().getSpanType()); + filter.doFilter(entry.getValue(), chainNode, costMap); + chainInfo.addNodes(chainNode); + } + chainInfo.generateChainToken(); + // HBaseUtil.saveCidTidMapping(key, chainInfo); + return chainInfo; + } - Map spanEntryMap = mergeSpanDataSet(spanList); - for (Map.Entry entry : spanEntryMap.entrySet()) { - ChainNode chainNode = new ChainNode(); - SpanNodeProcessFilter filter = SpanNodeProcessChain - .getProcessChainByCallType(entry.getValue().getSpanType()); - filter.doFilter(entry.getValue(), chainNode, costMap); - chainInfo.addNodes(chainNode); - } - chainInfo.generateChainToken(); - //HBaseUtil.saveCidTidMapping(key, chainInfo); - return chainInfo; - } - - private static Map mergeSpanDataSet(List spanList) { - Map spanEntryMap = new LinkedHashMap(); - for (int i = spanList.size() - 1; i >= 0; i--) { - Span span = spanList.get(i); - SpanEntry spanEntry = spanEntryMap.get(span.getParentLevel() + "." - + span.getLevelId()); - if (spanEntry == null) { - spanEntry = new SpanEntry(); - spanEntryMap.put( - span.getParentLevel() + "." + span.getLevelId(), - spanEntry); - } - spanEntry.setSpan(span); - } - return spanEntryMap; - } + private static Map mergeSpanDataSet(List spanList) { + Map spanEntryMap = new LinkedHashMap(); + for (int i = spanList.size() - 1; i >= 0; i--) { + Span span = spanList.get(i); + SpanEntry spanEntry = spanEntryMap.get(span.getParentLevel() + "." + + span.getLevelId()); + if (spanEntry == null) { + spanEntry = new SpanEntry(); + spanEntryMap.put( + span.getParentLevel() + "." + span.getLevelId(), + spanEntry); + } + spanEntry.setSpan(span); + } + return spanEntryMap; + } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java index b82492c39..3475baaf6 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java @@ -3,6 +3,9 @@ package com.ai.cloud.skywalking.analysis.chainbuild; import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainTree; import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo; import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; +import com.google.gson.Gson; +import com.google.gson.JsonParser; + import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; @@ -13,35 +16,42 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.Iterator; -public class ChainBuildReducer extends Reducer { +public class ChainBuildReducer extends Reducer { + private Logger logger = LoggerFactory.getLogger(ChainBuildReducer.class); - private Logger logger = LoggerFactory - .getLogger(ChainBuildReducer.class.getName()); + @Override + protected void setup(Context context) throws IOException, + InterruptedException { + ConfigInitializer.initialize(); + } - @Override - protected void setup(Context context) throws IOException, - InterruptedException { - ConfigInitializer.initialize(); - } + @Override + protected void reduce(Text key, Iterable values, Context context) + throws IOException, InterruptedException { + doReduceAction(Bytes.toString(key.getBytes()), values.iterator()); + } - @Override - protected void reduce(Text key, Iterable values, Context context) throws IOException, - InterruptedException { - doReduceAction(Bytes.toString(key.getBytes()), values.iterator()); - } + public void doReduceAction(String key, Iterator chainInfoIterator) + throws IOException, InterruptedException { + CallChainTree chainTree = CallChainTree.load(key); - public static void doReduceAction(String key, Iterator chainInfoIterator) throws IOException, InterruptedException { - CallChainTree chainTree = CallChainTree.load(key); + while (chainInfoIterator.hasNext()) { + String callChainData = chainInfoIterator.next().toString(); + ChainInfo chainInfo = null; + try { + chainInfo = new Gson().fromJson(callChainData, ChainInfo.class); + if (chainInfo.getChainStatus() == ChainInfo.ChainStatus.NORMAL) { + chainTree.processMerge(chainInfo); + } + // 合并数据 + chainTree.summary(chainInfo); + } catch (Exception e) { + logger.error( + "Failed to summary call chain, maybe illegal data:" + + callChainData, e); + } + } - while (chainInfoIterator.hasNext()) { - ChainInfo chainInfo = chainInfoIterator.next(); - if (chainInfo.getChainStatus() == ChainInfo.ChainStatus.NORMAL) { - chainTree.processMerge(chainInfo); - } - //合并数据 - chainTree.summary(chainInfo); - } - - chainTree.saveToHbase(); - } + chainTree.saveToHbase(); + } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/Tid2CidECovertException.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/Tid2CidECovertException.java new file mode 100644 index 000000000..6ccef0258 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/exception/Tid2CidECovertException.java @@ -0,0 +1,14 @@ +package com.ai.cloud.skywalking.analysis.chainbuild.exception; + +public class Tid2CidECovertException extends Exception{ + private static final long serialVersionUID = -4679233837335940374L; + + public Tid2CidECovertException(String msg){ + super(msg); + } + + public Tid2CidECovertException(String msg, Exception cause){ + super(msg, cause); + } + +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/ChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/ChainInfo.java index 69ae69da1..1e6b20217 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/ChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/ChainInfo.java @@ -1,64 +1,44 @@ package com.ai.cloud.skywalking.analysis.chainbuild.po; -import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator; -import com.google.gson.Gson; -import com.google.gson.JsonObject; -import com.google.gson.JsonParser; -import com.google.gson.reflect.TypeToken; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.io.Writable; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; +import java.io.Serializable; import java.util.ArrayList; import java.util.List; -public class ChainInfo implements Writable { - private String callEntrance; +import org.apache.hadoop.hbase.client.Put; + +import com.ai.cloud.skywalking.analysis.chainbuild.exception.Tid2CidECovertException; +import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator; + +public class ChainInfo implements Serializable { + private static final long serialVersionUID = -7194044877533469817L; + + /** + * 0节点的viewpoint，用于明文标识入口 + */ + private String callEntrance; private String cid; private ChainStatus chainStatus = ChainStatus.NORMAL; - private List nodes; + private List nodes = new ArrayList(); private String userId = null; private ChainNode firstChainNode; private long startDate; + private String tid; - public ChainInfo(String userId) { + public ChainInfo(String tid) { super(); - this.userId = userId; - } - - public ChainInfo() { - this.nodes = new ArrayList(); - } - - @Override - public void write(DataOutput out) throws IOException { - out.write(new Gson().toJson(this).getBytes()); - } - - @Override - public void readFields(DataInput in) throws IOException { - JsonObject jsonObject = (JsonObject) new JsonParser().parse(in - .readLine()); - cid = jsonObject.get("cid").getAsString(); - chainStatus = ChainStatus.convert(jsonObject.get("chainStatus") - .getAsCharacter()); - nodes = new Gson().fromJson(jsonObject.get("nodes"), - new TypeToken>() { - }.getType()); - userId = jsonObject.get("userId").getAsString(); + this.tid = tid; } public String getCID() { return cid; } - public String getEntranceNodeToken() { + public String getEntranceNodeToken() throws Tid2CidECovertException { if (firstChainNode == null) { - return ""; + throw new Tid2CidECovertException("tid[" + tid + "] can't find span node with level=0."); } else { - return firstChainNode.getNodeToken(); + return this.getUserId() + ":" + + firstChainNode.getNodeToken(); } } @@ -106,7 +86,7 @@ public class ChainInfo implements Writable { } public void saveToHBase(Put put) { - + //TODO： @zhangxin，未完成的入库代码 } public enum ChainStatus { diff --git a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java index 974755c31..5a4e19cbd 100644 --- a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java +++ b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java @@ -1,24 +1,31 @@ package com.ai.cloud.skywalking.analysis.mapper; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +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.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.io.Text; +import org.junit.Before; +import org.junit.Test; + import com.ai.cloud.skywalking.analysis.chainbuild.ChainBuildMapper; import com.ai.cloud.skywalking.analysis.chainbuild.ChainBuildReducer; import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo; import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; import com.ai.cloud.skywalking.analysis.config.HBaseTableMetaData; 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.List; +import com.google.gson.Gson; /** * Created by astraea on 2016/1/15. @@ -39,10 +46,10 @@ public class CallChainMapperTest { List spanList = selectByTraceId(chain_Id); ChainInfo chainInfo = ChainBuildMapper.spanToChainInfo(chain_Id, spanList); - List chainInfos = new ArrayList(); - chainInfos.add(chainInfo); + List chainInfos = new ArrayList(); + chainInfos.add(new Text(new Gson().toJson(chainInfo))); - ChainBuildReducer.doReduceAction(chainInfo.getCallEntrance(), chainInfos.iterator()); + new ChainBuildReducer().doReduceAction(chainInfo.getCallEntrance(), chainInfos.iterator()); } public static List selectByTraceId(String traceId) throws IOException {