kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. 修改MapReduce的过程，将以前的统计计算都放在一个Reduce里面做修改为将统计结果放在多处Reduce里面做 

2. 新新Span的排序类
This commit is contained in:
ascrutae 2016-05-02 18:27:47 +08:00
parent 89dbf1e5d6
commit fe2ef6a782
12 changed files with 369 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildMapper.java
View File

@ -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<Text, Text> {
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<Text, Text> {
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<Text, Text> {
public static ChainInfo spanToChainInfo(String tid, List<Span> spanList) {
SubLevelSpanCostCounter costMap = new SubLevelSpanCostCounter();
ChainInfo chainInfo = new ChainInfo(tid);
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);
}
});
Collections.sort(spanList, new SpanLevelIdComparators.SpanASCComparator());
Map<String, SpanEntry> spanEntryMap = mergeSpanDataSet(spanList);
for (Map.Entry<String, SpanEntry> entry : spanEntryMap.entrySet()) {

65
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java
View File

@ -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<Text, Text, Text, IntWritable> {
protected void reduce(Text key, Iterable<Text> 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<Text> 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<Text> 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<Text> 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);
// }
// }
}

11
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/ISummaryAction.java Normal file
View File

@ -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;
}

43
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/DateDetailSummaryAction.java Normal file
View File

@ -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;
}
}

31
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/action/impl/RelationShipAction.java Normal file
View File

@ -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();
}
}

20
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTree.java
View File

@ -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<String, CallChainTreeNode> getNodes() {
return nodes;
}
}

88
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTreeNode.java
View File

@ -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 {
* 按分钟维度进行汇总<br/>
* 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 {

12
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java
View File

@ -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<String, CallChainDetailForMysql> entry : callChainDetailMap.entrySet()) {
entry.getValue().saveToMysql();
}
@ -104,4 +99,5 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
}
HBaseUtil.batchSaveChainInfo(chainInfoPuts);
}
}

56
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SummaryType.java
View File

@ -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;
}
}

2
skywalking-analysis/src/main/resources/analysis.conf
View File

@ -17,4 +17,4 @@ chainnodesummary.interval=1
reducer.reducer_number=4
mapreduce.java_opts=-Xmx512m
mapreduce.java_opts=-Xmx768m

37
skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java
View File

@ -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<Span> spanList = selectByTraceId(chain_Id);
ChainInfo chainInfo = ChainBuildMapper.spanToChainInfo(chain_Id, spanList);
List<Text> 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<Text> chainInfos = new ArrayList<Text>();
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<Span> 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<String, Integer> entry : summaryResult.entrySet()){
for (Map.Entry<String, Integer> 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);

86
skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/SpanLevelIdComparators.java Normal file
View File

@ -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<Span> {
@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<Span> {
@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;
}
}