From 3ba2f2217f858831777d306600424217d4324555 Mon Sep 17 00:00:00 2001 From: zhangxin10 Date: Mon, 18 Jan 2016 16:17:50 +0800 Subject: [PATCH] =?UTF-8?q?1.=E5=AE=8C=E6=88=90Map=E9=83=A8=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analysis/mapper/CallChainMapper.java | 63 +++++++------- .../skywalking/analysis/model/ChainNode.java | 9 -- .../skywalking/analysis/util/HBaseUtil.java | 39 +-------- .../analysis/mapper/CallChainMapperTest.java | 82 +------------------ .../ai/cloud/skywalking/protocol/Span.java | 2 +- 5 files changed, 39 insertions(+), 156 deletions(-) diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java index ab9ca45db..801c013b4 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java @@ -1,20 +1,5 @@ package com.ai.cloud.skywalking.analysis.mapper; -import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import org.apache.hadoop.hbase.Cell; -import org.apache.hadoop.hbase.client.Result; -import org.apache.hadoop.hbase.io.ImmutableBytesWritable; -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.filter.SpanNodeProcessChain; import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; import com.ai.cloud.skywalking.analysis.model.ChainInfo; @@ -23,6 +8,17 @@ import com.ai.cloud.skywalking.analysis.model.CostMap; import com.ai.cloud.skywalking.analysis.model.SpanEntry; import com.ai.cloud.skywalking.analysis.util.HBaseUtil; import com.ai.cloud.skywalking.protocol.Span; +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 java.io.IOException; +import java.util.*; public class CallChainMapper extends TableMapper { private Logger logger = LoggerFactory.getLogger(CallChainMapper.class.getName()); @@ -30,35 +26,44 @@ public class CallChainMapper extends TableMapper { @Override protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException { - - ChainInfo chainInfo = new ChainInfo(); List spanList = new ArrayList(); - CostMap costMap = new CostMap(); for (Cell cell : value.rawCells()) { Span span = new Span(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())); spanList.add(span); } + ChainInfo chainInfo = spanToChainInfo(key.toString(), spanList); + context.write(new Text(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()), chainInfo); + } - //TODO: sort spanList - - Map spanEntryMap = mergeSpanDataset(spanList); + public static ChainInfo spanToChainInfo(String key, List spanList) { + CostMap costMap = new CostMap(); + 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.saveData(key.toString(), chainInfo); - context.write(new Text(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()), chainInfo); + chainInfo.generateChainToken(); + HBaseUtil.saveData(key, chainInfo); + return chainInfo; } - private Map mergeSpanDataset(List spanList) { - Map spanEntryMap = new HashMap(); - for (Span span : spanList) { + 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(); diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java index 14074d2a4..e18083e60 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java @@ -67,10 +67,6 @@ public class ChainNode { this.callType = callType; } - public String getCallType() { - return callType; - } - public void setUserId(String userId) { this.userId = userId; } @@ -79,10 +75,6 @@ public class ChainNode { return userId; } - public String getBusinessKey() { - return businessKey; - } - public void setBusinessKey(String businessKey) { this.businessKey = businessKey; } @@ -110,5 +102,4 @@ public class ChainNode { } } } - } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java index 7c24e65d0..21201b3ad 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java @@ -2,7 +2,6 @@ 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; @@ -32,41 +31,9 @@ public class HBaseUtil { 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())); - } - + put.addColumn(Bytes.toBytes(Config.HBase.TRACE_INFO_COLUMN_FAMILY), + Bytes.toBytes(Config.TraceInfo.TRACE_INFO_COLUMN_CID), + Bytes.toBytes(chainInfo.getChainToken())); try { table.put(put); if (logger.isDebugEnabled()) { 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 4a8515b6a..f4e3a6929 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 @@ -3,14 +3,7 @@ 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; @@ -23,9 +16,7 @@ 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. @@ -43,78 +34,7 @@ public class CallChainMapperTest { public void testMap() throws Exception { ConfigInitializer.initialize(); List spanList = selectByTraceId(chain_Id); - ChainInfo chainInfo = new ChainInfo(); - CostMap costMap = new CostMap(); - - 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.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 mergeSpanDataset(List spanList) { - Map spanEntryMap = new HashMap(); - for (Span span : spanList) { - 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; + ChainInfo chainInfo = CallChainMapper.spanToChainInfo(chain_Id, spanList); } public static List selectByTraceId(String traceId) throws IOException { diff --git a/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java index ea9a0de8d..cb6f74eae 100644 --- a/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java +++ b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/Span.java @@ -48,7 +48,7 @@ public class Span extends SpanData { processNo = fieldValues[12].trim(); applicationId = fieldValues[13].trim(); userId = fieldValues[14].trim(); - if (fieldValues.length > 14) { + if (fieldValues.length > 15) { callType = fieldValues[15].trim(); } this.originData = originData;