From fe2ef6a782d08df389f54cf597a2d09849a55f69 Mon Sep 17 00:00:00 2001 From: ascrutae Date: Mon, 2 May 2016 18:27:47 +0800 Subject: [PATCH] =?UTF-8?q?1.=20=E4=BF=AE=E6=94=B9MapReduce=E7=9A=84?= =?UTF-8?q?=E8=BF=87=E7=A8=8B=EF=BC=8C=E5=B0=86=E4=BB=A5=E5=89=8D=E7=9A=84?= =?UTF-8?q?=E7=BB=9F=E8=AE=A1=E8=AE=A1=E7=AE=97=E9=83=BD=E6=94=BE=E5=9C=A8?= =?UTF-8?q?=E4=B8=80=E4=B8=AAReduce=E9=87=8C=E9=9D=A2=E5=81=9A=E4=BF=AE?= =?UTF-8?q?=E6=94=B9=E4=B8=BA=E5=B0=86=E7=BB=9F=E8=AE=A1=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E6=94=BE=E5=9C=A8=E5=A4=9A=E5=A4=84Reduce=E9=87=8C=E9=9D=A2?= =?UTF-8?q?=E5=81=9A=202.=20=E6=96=B0=E6=96=B0Span=E7=9A=84=E6=8E=92?= =?UTF-8?q?=E5=BA=8F=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analysis/chainbuild/ChainBuildMapper.java | 47 ++++++---- .../chainbuild/ChainBuildReducer.java | 65 +++++++++----- .../chainbuild/action/ISummaryAction.java | 11 +++ .../action/impl/DateDetailSummaryAction.java | 43 +++++++++ .../action/impl/RelationShipAction.java | 31 +++++++ .../chainbuild/entity/CallChainTree.java | 20 ++--- .../chainbuild/entity/CallChainTreeNode.java | 88 ++++++++++--------- ...ficTimeCallTreeMergedChainIdContainer.java | 12 +-- .../analysis/chainbuild/po/SummaryType.java | 56 ++++++++---- .../src/main/resources/analysis.conf | 2 +- .../analysis/mapper/CallChainMapperTest.java | 37 ++++++-- .../util/SpanLevelIdComparators.java | 86 ++++++++++++++++++ 12 files changed, 369 insertions(+), 129 deletions(-) create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/ISummaryAction.java create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/DateDetailSummaryAction.java create mode 100644 skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/RelationShipAction.java create mode 100644 skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/SpanLevelIdComparators.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 5cdaa2941..98289bd17 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 @@ -7,9 +7,11 @@ 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.po.SummaryType; import com.ai.cloud.skywalking.analysis.chainbuild.util.SubLevelSpanCostCounter; +import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator; 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.ai.cloud.skywalking.util.SpanLevelIdComparators; import com.google.gson.Gson; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.client.Result; @@ -21,11 +23,16 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import java.io.IOException; +import java.text.SimpleDateFormat; import java.util.*; public class ChainBuildMapper extends TableMapper { private Logger logger = LogManager.getLogger(ChainBuildMapper.class); + private SimpleDateFormat hourSimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH"); + private SimpleDateFormat daySimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd"); + private SimpleDateFormat monthSimpleDateFormat = new SimpleDateFormat("yyyy-MM"); + private SimpleDateFormat yearSimpleDateFormat = new SimpleDateFormat("yyyy"); @Override protected void setup(Context context) throws IOException, @@ -57,14 +64,27 @@ public class ChainBuildMapper extends TableMapper { logger.debug("convert tid[" + Bytes.toString(key.get()) + "] to chain with cid[" + chainInfo.getCID() + "]."); if (chainInfo.getCallEntrance() != null && chainInfo.getCallEntrance().length() > 0) { - context.write( - new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.MINUTER.getValue()), new Text(new Gson().toJson(chainInfo))); - context.write( - new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.HOUR.getValue()), new Text(new Gson().toJson(chainInfo))); - context.write( - new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.DAY.getValue()), new Text(new Gson().toJson(chainInfo))); - context.write( - new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.MONTH.getValue()), new Text(new Gson().toJson(chainInfo))); + for (ChainNode chainNode : chainInfo.getNodes()) { + context.write(new Text(SummaryType.HOUR.getValue() + "-" + hourSimpleDateFormat.format( + new Date(chainNode.getStartDate()) + ) + ":" + chainInfo.getCallEntrance()), new Text(new Gson().toJson(chainNode))); + + context.write(new Text(SummaryType.DAY.getValue() + "-" + daySimpleDateFormat.format( + new Date(chainNode.getStartDate()) + ) + ":" + chainInfo.getCallEntrance()), new Text(new Gson().toJson(chainNode))); + + context.write(new Text(SummaryType.MONTH.getValue() + "-" + monthSimpleDateFormat.format( + new Date(chainNode.getStartDate()) + ) + ":" + chainInfo.getCallEntrance()), new Text(new Gson().toJson(chainNode))); + + context.write(new Text(SummaryType.YEAR.getValue() + "-" + yearSimpleDateFormat.format( + new Date(chainNode.getStartDate()) + ) + ":" + chainInfo.getCallEntrance()), new Text(new Gson().toJson(chainNode))); + } + // Reduce key : R-CallEntrance + context.write(new Text(SummaryType.RELATIONSHIP + "-" + TokenGenerator.generateTreeToken(chainInfo.getCallEntrance()) + + ":" + chainInfo.getCallEntrance()), + new Text(new Gson().toJson(chainInfo))); } } catch (Exception e) { logger.error("Failed to mapper call chain[" + key.toString() + "]", @@ -75,16 +95,7 @@ public class ChainBuildMapper extends TableMapper { 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); - } - }); + Collections.sort(spanList, new SpanLevelIdComparators.SpanASCComparator()); Map spanEntryMap = mergeSpanDataSet(spanList); for (Map.Entry entry : spanEntryMap.entrySet()) { 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 a8d146c6d..b3e06df32 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 @@ -1,12 +1,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.chainbuild.po.SpecificTimeCallTreeMergedChainIdContainer; +import com.ai.cloud.skywalking.analysis.chainbuild.action.ISummaryAction; import com.ai.cloud.skywalking.analysis.chainbuild.po.SummaryType; import com.ai.cloud.skywalking.analysis.config.Config; import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; -import com.google.gson.Gson; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; @@ -34,37 +31,61 @@ public class ChainBuildReducer extends Reducer { protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { String reduceKey = Bytes.toString(key.getBytes()); - int index = reduceKey.indexOf("@#!"); - if (index == -1){ + int index = reduceKey.indexOf(":"); + if (index == -1) { return; } - - doReduceAction(reduceKey.substring(0, index - 1), SummaryType.convert(reduceKey.substring(index + 3)), values.iterator()); + String summaryTypeAndDateStr = reduceKey.substring(0, index - 1); + String entryKey = reduceKey.substring(index + 1); + ISummaryAction summaryAction = SummaryType.chooseSummaryAction(summaryTypeAndDateStr, entryKey); + doReduceAction(summaryAction, values.iterator()); } - public void doReduceAction(String key, SummaryType summaryType, Iterator chainInfoIterator) - throws IOException, InterruptedException { - CallChainTree chainTree = CallChainTree.load(key); - SpecificTimeCallTreeMergedChainIdContainer container - = new SpecificTimeCallTreeMergedChainIdContainer(chainTree.getTreeToken()); - while (chainInfoIterator.hasNext()) { - String callChainData = chainInfoIterator.next().toString(); - ChainInfo chainInfo = null; + public void doReduceAction(ISummaryAction summaryAction, Iterator iterator) { + while (iterator.hasNext()) { + String summaryData = iterator.next().toString(); try { - chainInfo = new Gson().fromJson(callChainData, ChainInfo.class); - container.addMergedChainIfNotContain(chainInfo); - chainTree.summary(chainInfo, summaryType); + summaryAction.doAction(summaryData); } catch (Exception e) { logger.error( "Failed to summary call chain, maybe illegal data:" - + callChainData, e); + + summaryData, e); } } + try { - container.saveToHBase(summaryType); - chainTree.saveToHbase(); + summaryAction.doSave(); } catch (Exception e) { logger.error("Failed to save summaryresult/chainTree.", e); } } +// +// public void doReduceAction(String key, SummaryType summaryType, String summaryDateStr, Iterator chainInfoIterator) +// throws IOException, InterruptedException { +// CallChainTree chainTree = CallChainTree.load(key); +// SpecificTimeCallTreeMergedChainIdContainer container +// = new SpecificTimeCallTreeMergedChainIdContainer(chainTree.getTreeToken()); +// while (chainInfoIterator.hasNext()) { +// String chainNodeData = chainInfoIterator.next().toString(); +//// ChainInfo chainInfo = null; +// ChainNode chainNode = null; +// try { +//// chainInfo = new Gson().fromJson(callChainData, ChainInfo.class); +// chainNode = new Gson().fromJson(chainNodeData, ChainNode.class); +// // container.addMergedChainNodeIdIfNotContain(chainNode); +// //container.addMergedChainIfNotContain(chainInfo); +// //chainTree.summary(chainInfo, summaryType); +// } catch (Exception e) { +// logger.error( +// "Failed to summary call chain, maybe illegal data:" +// + chainNodeData, e); +// } +// } +// try { +// container.saveToHBase(summaryType); +// chainTree.saveToHBase(); +// } catch (Exception e) { +// logger.error("Failed to save summaryresult/chainTree.", e); +// } +// } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/ISummaryAction.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/ISummaryAction.java new file mode 100644 index 000000000..24d12a6a3 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/ISummaryAction.java @@ -0,0 +1,11 @@ +package com.ai.cloud.skywalking.analysis.chainbuild.action; + + +import java.io.IOException; +import java.sql.SQLException; + +public interface ISummaryAction { + void doAction(String summaryData) throws IOException; + + void doSave() throws InterruptedException, SQLException, IOException; +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/DateDetailSummaryAction.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/DateDetailSummaryAction.java new file mode 100644 index 000000000..3d32327d2 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/DateDetailSummaryAction.java @@ -0,0 +1,43 @@ +package com.ai.cloud.skywalking.analysis.chainbuild.action.impl; + +import com.ai.cloud.skywalking.analysis.chainbuild.action.ISummaryAction; +import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainTree; +import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainTreeNode; +import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainNode; +import com.ai.cloud.skywalking.analysis.chainbuild.po.SummaryType; +import com.google.gson.Gson; + +import java.io.IOException; +import java.sql.SQLException; + +public class DateDetailSummaryAction implements ISummaryAction { + private CallChainTree callChainTree; + private String summaryDate; + private SummaryType summaryType; + + public DateDetailSummaryAction(String entryKey, String summaryDate) throws IOException { + callChainTree = CallChainTree.load(entryKey); + this.summaryDate = summaryDate; + } + + @Override + public void doAction(String summaryData) throws IOException { + ChainNode chainNode = new Gson().fromJson(summaryData, ChainNode.class); + CallChainTreeNode newCallChainTreeNode = new CallChainTreeNode(chainNode); + CallChainTreeNode callChainTreeNode = callChainTree.getNodes().get(newCallChainTreeNode.getTreeNodeId()); + if (callChainTreeNode == null) { + callChainTreeNode = newCallChainTreeNode; + callChainTree.getNodes().put(newCallChainTreeNode.getTreeNodeId(), callChainTreeNode); + } + callChainTreeNode.summary(callChainTree.getTreeToken(), chainNode, summaryType, summaryDate); + } + + @Override + public void doSave() throws InterruptedException, SQLException, IOException { + callChainTree.saveToHBase(summaryType); + } + + public void setSummaryType(SummaryType summaryType) { + this.summaryType = summaryType; + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/RelationShipAction.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/RelationShipAction.java new file mode 100644 index 000000000..e5c112dc7 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/RelationShipAction.java @@ -0,0 +1,31 @@ +package com.ai.cloud.skywalking.analysis.chainbuild.action.impl; + +import com.ai.cloud.skywalking.analysis.chainbuild.action.ISummaryAction; +import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainTree; +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.po.SpecificTimeCallTreeMergedChainIdContainer; +import com.google.gson.Gson; + +import java.io.IOException; +import java.sql.SQLException; + +public class RelationShipAction implements ISummaryAction { + private CallChainTree chainTree; + private SpecificTimeCallTreeMergedChainIdContainer container; + public RelationShipAction(String entryKey) throws IOException { + chainTree = CallChainTree.load(entryKey); + container = new SpecificTimeCallTreeMergedChainIdContainer(chainTree.getTreeToken()); + } + + @Override + public void doAction(String summaryData) throws IOException { + ChainInfo chainInfo = new Gson().fromJson(summaryData, ChainInfo.class); + container.addMergedChainIfNotContain(chainInfo); + } + + @Override + public void doSave() throws InterruptedException, SQLException, IOException { + container.saveToHBase(); + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTree.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTree.java index 317a63c42..1ee88e2a0 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTree.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTree.java @@ -36,25 +36,17 @@ public class CallChainTree { return chain; } - public void summary(ChainInfo chainInfo, SummaryType summaryType) throws IOException { - for (ChainNode node : chainInfo.getNodes()) { - CallChainTreeNode newCallChainTreeNode = new CallChainTreeNode(node); - CallChainTreeNode callChainTreeNode = nodes.get(newCallChainTreeNode.getTreeNodeId()); - if (callChainTreeNode == null) { - callChainTreeNode = newCallChainTreeNode; - nodes.put(newCallChainTreeNode.getTreeNodeId(), callChainTreeNode); - } - callChainTreeNode.summary(treeToken, node, summaryType); - } - } - - public void saveToHbase() throws IOException, InterruptedException { + public void saveToHBase(SummaryType summaryType) throws IOException, InterruptedException { for (CallChainTreeNode entry : nodes.values()) { - entry.saveSummaryResultToHBase(); + entry.saveSummaryResultToHBase(summaryType); } } public String getTreeToken() { return treeToken; } + + public Map getNodes() { + return nodes; + } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTreeNode.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTreeNode.java index 0cebefa0a..2ff895796 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTreeNode.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTreeNode.java @@ -61,25 +61,32 @@ public class CallChainTreeNode { this.viewPointId = node.getViewPoint(); } - public void summary(String treeId, ChainNode node, SummaryType summaryType) throws IOException { + public void summary(String treeId, ChainNode node, SummaryType summaryType, String summaryDate) throws IOException { Calendar calendar = Calendar.getInstance(); calendar.setTime(new Date(node.getStartDate())); - if (summaryType == SummaryType.MINUTER) { - summaryMinResult(treeId, node, calendar); - } - if (summaryType == SummaryType.DAY) { - summaryHourResult(treeId, node, calendar); - } - if (summaryType == SummaryType.DAY) { - summaryDayResult(treeId, node, calendar); - } - if (summaryType == SummaryType.MONTH) { - summaryMonthResult(treeId, node, calendar); + switch (summaryType) { + case HOUR: { + summaryMinResult(treeId, node, calendar, summaryDate); + break; + } + case DAY: { + summaryHourResult(treeId, node, calendar, summaryDate); + break; + } + case MONTH: { + summaryDayResult(treeId, node, calendar, summaryDate); + break; + } + case YEAR: { + summaryMonthResult(treeId, node, calendar, summaryDate); + break; + } } + } - private void summaryMonthResult(String treeId, ChainNode node, Calendar calendar) throws IOException { - String keyOfMonthSummaryTable = generateKeyOfMonthSummaryTable(treeId, calendar); + private void summaryMonthResult(String treeId, ChainNode node, Calendar calendar, String summaryDate) throws IOException { + String keyOfMonthSummaryTable = generateRowKey(treeId, summaryDate); ChainNodeSpecificMonthSummary monthSummary = chainNodeSpecificMonthSummaryContainer.get(keyOfMonthSummaryTable); if (monthSummary == null) { if (Config.AnalysisServer.IS_ACCUMULATE_MODE) { @@ -92,8 +99,8 @@ public class CallChainTreeNode { monthSummary.summary(String.valueOf(calendar.get(Calendar.MONTH) + 1), node); } - private void summaryDayResult(String treeId, ChainNode node, Calendar calendar) throws IOException { - String keyOfDaySummaryTable = generateKeyOfDaySummaryTable(treeId, calendar); + private void summaryDayResult(String treeId, ChainNode node, Calendar calendar, String summaryDate) throws IOException { + String keyOfDaySummaryTable = generateRowKey(treeId, summaryDate); ChainNodeSpecificDaySummary daySummary = chainNodeSpecificDaySummaryContainer.get(keyOfDaySummaryTable); if (daySummary == null) { if (Config.AnalysisServer.IS_ACCUMULATE_MODE) { @@ -106,8 +113,8 @@ public class CallChainTreeNode { daySummary.summary(String.valueOf(calendar.get(Calendar.DAY_OF_MONTH)), node); } - private void summaryHourResult(String treeId, ChainNode node, Calendar calendar) throws IOException { - String keyOfHourSummaryTable = generateKeyOfHourSummaryTable(treeId, calendar); + private void summaryHourResult(String treeId, ChainNode node, Calendar calendar, String summaryDate) throws IOException { + String keyOfHourSummaryTable = generateRowKey(treeId, summaryDate); ChainNodeSpecificHourSummary hourSummary = chainNodeSpecificHourSummaryContainer.get(keyOfHourSummaryTable); if (hourSummary == null) { if (Config.AnalysisServer.IS_ACCUMULATE_MODE) { @@ -124,8 +131,8 @@ public class CallChainTreeNode { * 按分钟维度进行汇总
* chainNodeContainer以treeId和时间（精确到分钟）为key，value为当前时间范围内的所有分钟的汇总数据 */ - private void summaryMinResult(String treeId, ChainNode node, Calendar calendar) throws IOException { - String keyOfMinSummaryTable = generateKeyOfMinSummaryTable(treeId, calendar); + private void summaryMinResult(String treeId, ChainNode node, Calendar calendar, String summaryDate) throws IOException { + String keyOfMinSummaryTable = generateRowKey(treeId, summaryDate); ChainNodeSpecificMinSummary minSummary = chainNodeSpecificMinSummaryContainer.get(keyOfMinSummaryTable); if (minSummary == null) { if (Config.AnalysisServer.IS_ACCUMULATE_MODE) { @@ -138,23 +145,10 @@ public class CallChainTreeNode { minSummary.summary(String.valueOf(calendar.get(Calendar.MINUTE)), node); } - private String generateKeyOfMonthSummaryTable(String treeId, Calendar calendar) { - return treeId + "/" + calendar.get(Calendar.YEAR); + private String generateRowKey(String treeId, String dateKey) { + return treeId + "/" + dateKey; } - private String generateKeyOfDaySummaryTable(String treeId, Calendar calendar) { - return treeId + "/" + calendar.get(Calendar.YEAR) + "-" + (calendar.get(Calendar.MONTH) + 1); - } - - private String generateKeyOfHourSummaryTable(String treeId, Calendar calendar) { - return treeId + "/" + calendar.get(Calendar.YEAR) + "-" + (calendar.get(Calendar.MONTH) + 1) + "-" - + calendar.get(Calendar.DAY_OF_MONTH); - } - - private String generateKeyOfMinSummaryTable(String treeId, Calendar calendar) { - return treeId + "/" + calendar.get(Calendar.YEAR) + "-" + (calendar.get(Calendar.MONTH) + 1) + "-" - + calendar.get(Calendar.DAY_OF_MONTH) + " " + calendar.get(Calendar.HOUR) + ":00:00"; - } @Override public String toString() { @@ -170,11 +164,25 @@ public class CallChainTreeNode { * @throws IOException * @throws InterruptedException */ - public void saveSummaryResultToHBase() throws IOException, InterruptedException { - batchSaveMinSummaryResult(); - batchSaveHourSummaryResult(); - batchSaveDaySummaryResult(); - batchSaveMonthSummaryResult(); + public void saveSummaryResultToHBase(SummaryType summaryType) throws IOException, InterruptedException { + switch (summaryType) { + case HOUR: { + batchSaveMinSummaryResult(); + break; + } + case DAY: { + batchSaveHourSummaryResult(); + break; + } + case MONTH: { + batchSaveDaySummaryResult(); + break; + } + case YEAR: { + batchSaveMonthSummaryResult(); + break; + } + } } private void batchSaveMonthSummaryResult() throws IOException, InterruptedException { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java index 4ad3dc400..941242dcb 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java @@ -44,8 +44,6 @@ public class SpecificTimeCallTreeMergedChainIdContainer { if (chainInfo.getChainStatus() == ChainInfo.ChainStatus.NORMAL) { callChainDetailMap.put(chainInfo.getCID(), new CallChainDetailForMysql(chainInfo, treeToken)); } - } else { - } } @@ -56,16 +54,13 @@ public class SpecificTimeCallTreeMergedChainIdContainer { return treeToken + "@" + calendar.get(Calendar.YEAR) + "-" + calendar.get(Calendar.MONTH); } - public void saveToHBase(SummaryType summaryType) throws IOException, InterruptedException, SQLException { + public void saveToHBase() throws IOException, InterruptedException, SQLException { batchSaveCurrentHasBeenMergedChainInfo(); batchSaveMergedChainId(); - batchSaveToMysql(summaryType); + batchSaveToMysql(); } - private void batchSaveToMysql(SummaryType summaryType) throws SQLException { - if (summaryType != SummaryType.MONTH) { - return; - } + private void batchSaveToMysql() throws SQLException { for (Map.Entry entry : callChainDetailMap.entrySet()) { entry.getValue().saveToMysql(); } @@ -104,4 +99,5 @@ public class SpecificTimeCallTreeMergedChainIdContainer { } HBaseUtil.batchSaveChainInfo(chainInfoPuts); } + } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SummaryType.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SummaryType.java index 0b4d6a3ac..5559f5eaa 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SummaryType.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SummaryType.java @@ -1,7 +1,13 @@ package com.ai.cloud.skywalking.analysis.chainbuild.po; +import com.ai.cloud.skywalking.analysis.chainbuild.action.ISummaryAction; +import com.ai.cloud.skywalking.analysis.chainbuild.action.impl.DateDetailSummaryAction; +import com.ai.cloud.skywalking.analysis.chainbuild.action.impl.RelationShipAction; + +import java.io.IOException; + public enum SummaryType { - MINUTER('m'), HOUR('H'), DAY('D'), MONTH('M'); + HOUR('H'), DAY('D'), MONTH('M'), YEAR('Y'), RELATIONSHIP('R'); private char value; @@ -9,23 +15,39 @@ public enum SummaryType { this.value = value; } - public static SummaryType convert(String value) { - char valueChar = value.charAt(0); - switch (valueChar) { - case 'm': - return MINUTER; - case 'H': - return HOUR; - case 'D': - return DAY; - case 'M': - return MONTH; - default: - throw new RuntimeException("Can not find the summary type[" + value + "]"); - } - } - public char getValue() { return value; } + + public static ISummaryAction chooseSummaryAction(String summaryTypeAndDateStr, String entryKey) throws IOException { + char valueChar = summaryTypeAndDateStr.charAt(0); + // HOUR : 2016-05-02/12 + // DAY : 2016-05-02 + // MONTH : 2016-05 + // YEAR : 2016 + // RELATIONSHIP : treeId + String summaryDateStr = summaryTypeAndDateStr.substring(summaryTypeAndDateStr.indexOf("-") + 1); + SummaryType type = null; + switch (valueChar) { + case 'H': + type = HOUR; + break; + case 'D': + type = DAY; + break; + case 'M': + type = MONTH; + break; + case 'Y': + type = YEAR; + break; + case 'R': + return new RelationShipAction(entryKey); + default: + throw new RuntimeException("Can not find the summary type[" + valueChar + "]"); + } + DateDetailSummaryAction summaryAction = new DateDetailSummaryAction(entryKey, summaryDateStr); + summaryAction.setSummaryType(type); + return summaryAction; + } } diff --git a/skywalking-analysis/src/main/resources/analysis.conf b/skywalking-analysis/src/main/resources/analysis.conf index 2ece0b01e..0cb7ba204 100644 --- a/skywalking-analysis/src/main/resources/analysis.conf +++ b/skywalking-analysis/src/main/resources/analysis.conf @@ -17,4 +17,4 @@ chainnodesummary.interval=1 reducer.reducer_number=4 -mapreduce.java_opts=-Xmx512m \ No newline at end of file +mapreduce.java_opts=-Xmx768m \ No newline at end of file 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 d80afbc14..a58ded1dd 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,7 +3,10 @@ package com.ai.cloud.skywalking.analysis.mapper; import com.ai.cloud.skywalking.analysis.chainbuild.ChainBuildMapper; import com.ai.cloud.skywalking.analysis.chainbuild.ChainBuildReducer; +import com.ai.cloud.skywalking.analysis.chainbuild.action.ISummaryAction; import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo; +import com.ai.cloud.skywalking.analysis.chainbuild.po.SummaryType; +import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator; import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; import com.ai.cloud.skywalking.analysis.config.HBaseTableMetaData; import com.ai.cloud.skywalking.protocol.Span; @@ -30,21 +33,38 @@ public class CallChainMapperTest { private static String ZK_QUORUM = "10.1.235.197,10.1.235.198,10.1.235.199"; // private static String ZK_QUORUM = "10.1.241.18,10.1.241.19,10.1.241.20"; private static String ZK_CLIENT_PORT = "29181"; - private static String chain_Id = "1.0a2.1457436000002.860568c.21818.49.82"; -// private static String chain_Id = "1.0a2.1453429608422.2701d43.6468.56.1"; + private static String chain_Id = "1.0a2.1453123911750.b3d7400.13582.96.2"; + //private static String chain_Id = "1.0a2.1453429608422.2701d43.6468.56.1"; private static Configuration configuration = null; private static Connection connection; + + private static SimpleDateFormat hourSimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH"); + private static SimpleDateFormat daySimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd"); + private static SimpleDateFormat monthSimpleDateFormat = new SimpleDateFormat("yyyy-MM"); + private static SimpleDateFormat yearSimpleDateFormat = new SimpleDateFormat("yyyy"); + public static void main(String[] args) throws Exception { ConfigInitializer.initialize(); + initHBaseClient(); List spanList = selectByTraceId(chain_Id); ChainInfo chainInfo = ChainBuildMapper.spanToChainInfo(chain_Id, spanList); + List chainNodeInfo = new ArrayList<>(); + chainNodeInfo.add(new Text(new Gson().toJson(chainInfo))); + Text key = new Text(SummaryType.RELATIONSHIP + "-" + TokenGenerator.generateTreeToken(chainInfo.getCallEntrance()) + + ":" + chainInfo.getCallEntrance()); + String reduceKey = Bytes.toString(key.getBytes()); + int index = reduceKey.indexOf(":"); + if (index == -1) { + return; + } - List chainInfos = new ArrayList(); - chainInfos.add(new Text(new Gson().toJson(chainInfo))); + String summaryTypeAndDateStr = reduceKey.substring(0, index - 1); + String entryKey = reduceKey.substring(index + 1); + ISummaryAction summaryAction = SummaryType.chooseSummaryAction(summaryTypeAndDateStr, entryKey); - new ChainBuildReducer().doReduceAction(chainInfo.getCallEntrance(), chainInfos.iterator()); + new ChainBuildReducer().doReduceAction(summaryAction, chainNodeInfo.iterator()); } public static List selectByTraceId(String traceId) throws IOException { @@ -68,13 +88,13 @@ public class CallChainMapperTest { for (ChainInfo chainInfo : chainInfoList) { String key = generateKey(chainInfo.getStartDate()); Integer total = summaryResult.get(key); - if (total == null){ + if (total == null) { total = 0; } summaryResult.put(key, ++total); } - for (Map.Entry entry : summaryResult.entrySet()){ + for (Map.Entry entry : summaryResult.entrySet()) { System.out.println(entry.getKey() + " " + entry.getValue()); } @@ -116,8 +136,7 @@ public class CallChainMapperTest { return chainInfos; } - - public void initHBaseClient() throws IOException { + public static void initHBaseClient() throws IOException { if (configuration == null) { configuration = HBaseConfiguration.create(); configuration.set("hbase.zookeeper.quorum", ZK_QUORUM); diff --git a/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/SpanLevelIdComparators.java b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/SpanLevelIdComparators.java new file mode 100644 index 000000000..ced973aec --- /dev/null +++ b/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/SpanLevelIdComparators.java @@ -0,0 +1,86 @@ +package com.ai.cloud.skywalking.util; + + +import com.ai.cloud.skywalking.protocol.Span; + +import java.util.Comparator; + +public class SpanLevelIdComparators { + + public static class SpanDESCComparator implements Comparator { + @Override + public int compare(Span span1, Span span2) { + String span1TraceLevel = getTraceLevelId(span1); + String span2TraceLevel = getTraceLevelId(span2); + return descComparator(span1TraceLevel, span2TraceLevel); + } + } + + public static class SpanASCComparator implements Comparator { + @Override + public int compare(Span span1, Span span2) { + String span1TraceLevel = getTraceLevelId(span1); + String span2TraceLevel = getTraceLevelId(span2); + return ascCompare(span1TraceLevel, span2TraceLevel); + } + } + + private static String getTraceLevelId(Span span) { + String spanTraceLevelId = null; + if (span.getParentLevel() == null || span.getParentLevel().length() == 0) { + spanTraceLevelId = span.getLevelId() + ""; + } else { + spanTraceLevelId = span.getParentLevel() + "." + span.getLevelId(); + } + return spanTraceLevelId; + } + + private static int descComparator(String levelId0, String levelId1) { + String[] levelId0Array = levelId0.split("\\."); + String[] levelId1Array = levelId1.split("\\."); + int result = -1; + int index = 0; + while (true) { + if (index >= levelId0Array.length) { + result = 1; + break; + } + + if (index >= levelId1Array.length) { + result = -1; + break; + } + result = -1 * new Integer(levelId0Array[index]).compareTo(new Integer(levelId1Array[index])); + if (result != 0) + break; + index++; + + } + return result; + } + + + private static int ascCompare(String levelId0, String levelId1) { + String[] levelId0Array = levelId0.split("\\."); + String[] levelId1Array = levelId1.split("\\."); + int result = -1; + int index = 0; + while (true) { + if (index >= levelId0Array.length) { + result = -1; + break; + } + + if (index >= levelId1Array.length) { + result = 1; + break; + } + result = new Integer(levelId0Array[index]).compareTo(new Integer(levelId1Array[index])); + if (result != 0) + break; + index++; + + } + return result; + } +}