diff --git a/skywalking-analysis/pom.xml b/skywalking-analysis/pom.xml index c74aeb101..d0c82b291 100644 --- a/skywalking-analysis/pom.xml +++ b/skywalking-analysis/pom.xml @@ -41,6 +41,12 @@ skywalking-protocol 1.0-SNAPSHOT + + junit + junit + 4.12 + test + diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java index 651b60d04..d32049fa7 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java @@ -41,7 +41,7 @@ public class AnalysisServerDriver extends Configured implements Tool { job.setJarByClass(AnalysisServerDriver.class); Scan scan = buildHBaseScan(args); - TableMapReduceUtil.initTableMapperJob(Config.HBase.TABLE_NAME, scan, CallChainMapper.class, + TableMapReduceUtil.initTableMapperJob(Config.HBase.CALL_CHAIN_TABLE_NAME, scan, CallChainMapper.class, String.class, ChainInfo.class, job); //TableMapReduceUtil.initTableReducerJob("sw-call-chain-model", CallChainReducer.class, job); return job.waitForCompletion(true) ? 0 : 1; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java index 9dbff39a4..2c3e81434 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java @@ -2,6 +2,38 @@ package com.ai.cloud.skywalking.analysis.config; public class Config { public static class HBase { - public static String TABLE_NAME; + + public static String TRACE_INFO_COLUMN_FAMILY = "trace_info"; + + public static String CALL_CHAIN_TABLE_NAME; + + public static String ZK_QUORUM; + + public static String ZK_CLIENT_PORT; + + public static String TRACE_INFO_TABLE_NAME; + + + } + + public static class TraceInfo { + + public static String PARENT_LEVEL_ID = "parentLevelId"; + + public static String LEVEL_ID = "levelId"; + + public static String BUSINESS_KEY = "businessKey"; + + public static String COST = "cost"; + + public static String TRACE_INFO_COLUMN_CID = "cid"; + + public static String STATUS = "status"; + + public static String USER_ID = "UId"; + } + + public static class Filter { + public static String FILTER_PACKAGE_NAME; } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ViewPointFilterFactory.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ViewPointFilterFactory.java deleted file mode 100644 index ea8b74c76..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ViewPointFilterFactory.java +++ /dev/null @@ -1,86 +0,0 @@ -package com.ai.cloud.skywalking.analysis.config; - -import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.util.HashMap; -import java.util.Map; -import java.util.Properties; -import java.util.Set; - -public class ViewPointFilterFactory { - - private static Logger logger = LoggerFactory.getLogger(ViewPointFilterFactory.class.getName()); - - private static Map filterType; - - private static Object lock = new Object(); - - private ViewPointFilterFactory() { - //Non - } - - private static void initFilterChain() { - Properties properties = new Properties(); - try { - properties.load(ViewPointFilterFactory.class.getResourceAsStream("/viewpointfilter.conf")); - } catch (IOException e) { - logger.error("Failed to found the conf file[viewpointfilter.conf]", e); - System.exit(-1); - } - - Set> entries = properties.entrySet(); - - for (Map.Entry entry : entries) { - String types = (String) entry.getKey(); - String filters = (String) entry.getValue(); - - String[] filterClasses = filters.split(">"); - ViewPointFilter filter = null; - for (int i = filterClasses.length - 1; i >= 0; i--) { - try { - Class filterClass = Class.forName(filterClasses[i]); - ViewPointFilter newFilter = (ViewPointFilter) filterClass.newInstance(); - newFilter.setViewPointFilter(filter); - filter = newFilter; - } catch (ClassNotFoundException e) { - logger.error("Failed to found class[" + filterClasses[i] + "].", e); - System.exit(-1); - } catch (InstantiationException e) { - logger.error("Failed to instance class[" + filterClasses[i] + "].", e); - System.exit(-1); - } catch (IllegalAccessException e) { - logger.error("Failed to access class[" + filterClasses[i] + "].", e); - System.exit(-1); - } - } - - - String[] type = types.split(","); - for (int i = 0; i < type.length; i++) { - filterType.put(type[i], filter); - } - } - } - - public static ViewPointFilter getFilter(String type) { - if (filterType == null) { - synchronized (lock) { - if (filterType == null) { - filterType = new HashMap(); - initFilterChain(); - } - } - } - - ViewPointFilter filter = filterType.get(type); - if (filter == null) { - throw new RuntimeException("Failed to found the filter[" + type + "]"); - } - - return filter; - } - -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/ChainNodeFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/ChainNodeFilter.java deleted file mode 100644 index cb5db2a09..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/ChainNodeFilter.java +++ /dev/null @@ -1,9 +0,0 @@ -package com.ai.cloud.skywalking.analysis.filter; - -import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.protocol.Span; - -public interface ChainNodeFilter { - void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain); -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/NodeChain.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/NodeChain.java deleted file mode 100644 index b30eee1db..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/NodeChain.java +++ /dev/null @@ -1,49 +0,0 @@ -package com.ai.cloud.skywalking.analysis.filter; - -import com.ai.cloud.skywalking.analysis.filter.impl.CopyAttrNodeFilter; -import com.ai.cloud.skywalking.analysis.filter.impl.CostNodeFilter; -import com.ai.cloud.skywalking.analysis.filter.impl.ViewPointNodeFilter; -import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.protocol.Span; - -import java.util.ArrayList; -import java.util.List; - -public class NodeChain { - - private int stage = 0; - - private static List filters = null; - - public void doChain(Span span, ChainNode chainNode, CostMap costMap) { - if (filters == null) { - filters = new ArrayList(); - init(); - } - - int subscript = stage; - stage = stage + 1; - if (subscript < filters.size()) { - filters.get(subscript).doFilter(span, chainNode, costMap, this); - } - } - - private static void init() { - filters.add(new CostNodeFilter()); - filters.add(new CopyAttrNodeFilter()); - filters.add(new ViewPointNodeFilter()); - } - - public static void main(String[] args) { - Span span = new Span("1.0a2.1452649597690.035d27f.1608.56.1@~ @~0@~" + - "http://localhost:8080/skywalking-web/order/save@~1452649597692@~5202@~" + - "astraea-PC/192.168.1.102@~0@~ @~W@~true@~ @~1608@~web-application@~6@~A "); - NodeChain nodeChain = new NodeChain(); - CostMap costMap = new CostMap(); - ChainNode chainNode = new ChainNode(); - nodeChain.doChain(span, chainNode, costMap); - - System.out.print("xx"); - } -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessChain.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessChain.java new file mode 100644 index 000000000..5cc728c12 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessChain.java @@ -0,0 +1,87 @@ +package com.ai.cloud.skywalking.analysis.filter; + +import com.ai.cloud.skywalking.analysis.config.Config; +import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; +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.protocol.Span; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; + +public class SpanNodeProcessChain { + private static Logger logger = LoggerFactory.getLogger(SpanNodeProcessChain.class.getName()); + private static Map filterMap; + private static Object lock = new Object(); + + private SpanNodeProcessChain() { + //Non + } + + private static void initFilterMap(Map filterMap) { + Properties properties = new Properties(); + + try { + properties.load(SpanNodeProcessChain.class.getResourceAsStream("/viewpointfilter.conf")); + } catch (IOException e) { + logger.error("Failed to find config file[viewpointfilter.conf]", e); + System.exit(-1); + } + + for (Map.Entry entry : properties.entrySet()) { + String[] filters = ((String) entry.getValue()).split("->"); + String[] types = ((String) entry.getKey()).split(","); + + SpanNodeProcessFilter currentFilter = null; + for (int i = filters.length - 1; i >= 0; i--) { + try { + Class filterClass = Class.forName(Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i]); + SpanNodeProcessFilter tmpSpanNodeFilter = (SpanNodeProcessFilter) filterClass.newInstance(); + tmpSpanNodeFilter.setNextProcessChain(currentFilter); + currentFilter = tmpSpanNodeFilter; + } catch (ClassNotFoundException e) { + logger.error("Filed to find class[" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "]", e); + System.exit(-1); + } catch (InstantiationException e) { + logger.error("Can not instance class[" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "]", e); + System.exit(-1); + } catch (IllegalAccessException e) { + logger.error("Can not access class[" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "]", e); + System.exit(-1); + } catch (ClassCastException e) { + logger.error("Class [" + Config.Filter.FILTER_PACKAGE_NAME + "." + filters[i] + "] is not subclass of " + + "com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter", e); + System.exit(-1); + } + } + + for (String type : types) { + filterMap.put(type, currentFilter); + } + + } + } + + public static SpanNodeProcessFilter getProcessChainByCallType(String callType) { + if (filterMap == null) { + synchronized (lock) { + if (filterMap == null) { + filterMap = new HashMap(); + initFilterMap(filterMap); + } + } + } + + if (filterMap.containsKey(callType)) { + return filterMap.get(callType); + } + + return filterMap.get("default"); + } + +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java new file mode 100644 index 000000000..74d5f5d6c --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/SpanNodeProcessFilter.java @@ -0,0 +1,21 @@ +package com.ai.cloud.skywalking.analysis.filter; + +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.protocol.Span; + +public abstract class SpanNodeProcessFilter { + + private SpanNodeProcessFilter nextProcessChain; + + public abstract void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap); + + public SpanNodeProcessFilter getNextProcessChain() { + return nextProcessChain; + } + + public void setNextProcessChain(SpanNodeProcessFilter nextProcessChain) { + this.nextProcessChain = nextProcessChain; + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java new file mode 100644 index 000000000..c02779bd4 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/AppendBusinessKeyFilter.java @@ -0,0 +1,18 @@ +package com.ai.cloud.skywalking.analysis.filter.impl; + +import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; +import com.ai.cloud.skywalking.analysis.model.ChainNode; +import com.ai.cloud.skywalking.analysis.model.CostMap; +import com.ai.cloud.skywalking.analysis.model.SpanEntry; + +public class AppendBusinessKeyFilter extends SpanNodeProcessFilter { + + @Override + public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) { + node.setViewPoint(node.getViewPoint() + spanEntry.getBusinessKey()); + + if (getNextProcessChain() != null) { + getNextProcessChain().doFilter(spanEntry, node, costMap); + } + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java new file mode 100644 index 000000000..a17a2d8ed --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrFilter.java @@ -0,0 +1,23 @@ +package com.ai.cloud.skywalking.analysis.filter.impl; + +import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; +import com.ai.cloud.skywalking.analysis.model.ChainNode; +import com.ai.cloud.skywalking.analysis.model.CostMap; +import com.ai.cloud.skywalking.analysis.model.SpanEntry; + +public class CopyAttrFilter extends SpanNodeProcessFilter { + + @Override + public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) { + node.setCallType(spanEntry.getCallType().toString()); + node.setStatus(spanEntry.getSpanStatus()); + node.setLevelId(spanEntry.getLevelId()); + node.setParentLevelId(spanEntry.getParentLevelId()); + node.setViewPoint(spanEntry.getViewPoint()); + node.setUserId(spanEntry.getUserId()); + + if (getNextProcessChain() != null) { + getNextProcessChain().doFilter(spanEntry, node, costMap); + } + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrNodeFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrNodeFilter.java deleted file mode 100644 index ed7e225fb..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrNodeFilter.java +++ /dev/null @@ -1,26 +0,0 @@ -package com.ai.cloud.skywalking.analysis.filter.impl; - -import com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter; -import com.ai.cloud.skywalking.analysis.filter.NodeChain; -import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.protocol.Span; - -public class CopyAttrNodeFilter implements ChainNodeFilter { - - @Override - public void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain) { - node.setCost(span.getCost()); - node.setLevelId(span.getLevelId()); - node.setParentLevelId(span.getParentLevel()); - node.setCallType(span.getCallType()); - - if (span.getExceptionStack() == null || span.getExceptionStack().length() == 0) { - node.setStatus(ChainNode.NodeStatus.NORMAL); - } else { - node.setStatus(ChainNode.NodeStatus.ABNORMAL); - } - - chain.doChain(span, node, costMap); - } -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CostNodeFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CostNodeFilter.java deleted file mode 100644 index 7337cce0e..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CostNodeFilter.java +++ /dev/null @@ -1,34 +0,0 @@ -package com.ai.cloud.skywalking.analysis.filter.impl; - -import com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter; -import com.ai.cloud.skywalking.analysis.filter.NodeChain; -import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.protocol.Span; - -public class CostNodeFilter implements ChainNodeFilter { - - @Override - public void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain) { - if (span.isReceiver()) { - costMap.put(span.getParentLevel() + "." + span.getLevelId() + "-S", span.getCost()); - - if (isFirstNode(span)) { - chain.doChain(span, node, costMap); - } - } else { - if (costMap.exists(span.getParentLevel())) { - costMap.put(span.getParentLevel(), costMap.get(span.getParentLevel()) + span.getCost()); - } else { - costMap.put(span.getParentLevel(), span.getCost()); - } - chain.doChain(span, node, costMap); - } - - - } - - private boolean isFirstNode(Span span) { - return span.getParentLevel().length() == 0 && span.getLevelId() == 0; - } -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java new file mode 100644 index 000000000..d0677e12c --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ProcessCostTimeFilter.java @@ -0,0 +1,24 @@ +package com.ai.cloud.skywalking.analysis.filter.impl; + +import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; +import com.ai.cloud.skywalking.analysis.model.ChainNode; +import com.ai.cloud.skywalking.analysis.model.CostMap; +import com.ai.cloud.skywalking.analysis.model.SpanEntry; + +public class ProcessCostTimeFilter extends SpanNodeProcessFilter { + @Override + public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) { + long subNodeCost = spanEntry.getCost(); + node.setCost(spanEntry.getCost()); + + if (costMap.exists(spanEntry.getParentLevelId())) { + subNodeCost += costMap.get(spanEntry.getParentLevelId()); + } + + costMap.put(spanEntry.getParentLevelId(), subNodeCost); + + if (getNextProcessChain() != null) { + getNextProcessChain().doFilter(spanEntry, node, costMap); + } + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java new file mode 100644 index 000000000..ad69d2759 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ReplaceAddressFilter.java @@ -0,0 +1,23 @@ +package com.ai.cloud.skywalking.analysis.filter.impl; + +import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; +import com.ai.cloud.skywalking.analysis.model.ChainNode; +import com.ai.cloud.skywalking.analysis.model.CostMap; +import com.ai.cloud.skywalking.analysis.model.SpanEntry; + +public class ReplaceAddressFilter extends SpanNodeProcessFilter { + + //ip:port regex + private static String IP_PORT_REGEX = "(([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))|localhost):([1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5]|[0-9]{1,4})"; + + @Override + public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) { + + String viewPoint = spanEntry.getViewPoint().replaceAll(IP_PORT_REGEX, spanEntry.getApplicationId()); + node.setViewPoint(viewPoint); + + if (getNextProcessChain() != null) { + getNextProcessChain().doFilter(spanEntry, node, costMap); + } + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java new file mode 100644 index 000000000..29ca49ba0 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/TokenGenerateFilter.java @@ -0,0 +1,23 @@ +package com.ai.cloud.skywalking.analysis.filter.impl; + +import com.ai.cloud.skywalking.analysis.filter.SpanNodeProcessFilter; +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.TokenGenerator; + +public class TokenGenerateFilter extends SpanNodeProcessFilter { + + @Override + public void doFilter(SpanEntry spanEntry, ChainNode node, CostMap costMap) { + + String nodeToken = TokenGenerator.generate(node.getParentLevelId() + "." + node.getLevelId() + + "-" + node.getViewPoint()); + + node.setNodeToken(nodeToken); + + if (getNextProcessChain() != null) { + getNextProcessChain().doFilter(spanEntry, node, costMap); + } + } +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ViewPointNodeFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ViewPointNodeFilter.java deleted file mode 100644 index 93fccaad8..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ViewPointNodeFilter.java +++ /dev/null @@ -1,20 +0,0 @@ -package com.ai.cloud.skywalking.analysis.filter.impl; - -import com.ai.cloud.skywalking.analysis.config.ViewPointFilterFactory; -import com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter; -import com.ai.cloud.skywalking.analysis.filter.NodeChain; -import com.ai.cloud.skywalking.analysis.model.ChainNode; -import com.ai.cloud.skywalking.analysis.model.CostMap; -import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter; -import com.ai.cloud.skywalking.protocol.Span; - -public class ViewPointNodeFilter implements ChainNodeFilter { - @Override - public void doFilter(Span span, ChainNode node, CostMap costMap, NodeChain chain) { - ViewPointFilter viewPointFilter = ViewPointFilterFactory.getFilter(span.getSpanType()); - String viewPoint = span.getViewPointId(); - viewPointFilter.doFilter(span, viewPoint); - node.setViewPoint(viewPoint); - chain.doChain(span, node, costMap); - } -} 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 b83fee4b7..48a3285ee 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,10 +1,13 @@ package com.ai.cloud.skywalking.analysis.mapper; -import com.ai.cloud.skywalking.analysis.filter.NodeChain; +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.protocol.CallType; +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.hbase.Cell; import org.apache.hadoop.hbase.client.Result; @@ -16,6 +19,10 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; public class CallChainMapper extends TableMapper { private Logger logger = LoggerFactory.getLogger(CallChainMapper.class.getName()); @@ -23,38 +30,93 @@ public class CallChainMapper extends TableMapper { @Override protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException { - CostMap costMap = new CostMap(); - ChainInfo chainInfo = new ChainInfo(); + ChainInfo chainInfo = new ChainInfo(); + List spanList = new ArrayList(); + CostMap costMap = new CostMap(); for (Cell cell : value.rawCells()) { - ChainNode node = new ChainNode(); Span span = new Span(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())); - new NodeChain().doChain(span, node, costMap); - chainInfo.getNodes().add(node); + spanList.add(span); } + 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()) { - CallType callType = CallType.valueOf(node.getCallType()); - switch (callType) { - case ASYNC: + if ((node.getParentLevelId() == null || node.getParentLevelId().length() == 0) + && node.getLevelId() == 0) { + firstNodeToken = node.getNodeToken(); + chainInfo.setUserId(node.getUserId()); + } - break; - case SYNC: + // 状态轮询 + if (node.getStatus() == ChainNode.NodeStatus.ABNORMAL) { + status = false; + } - break; - case LOCAL: + // 设置时间 + computeChainNodeCost(costMap, node); - break; + 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(key.toString(), chainInfo); + + context.write(new Text(key.toString() + ":" + firstNodeToken), 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 && span.isReceiver()) { + if (span.isReceiver()) { + spanEntry.setServerSpan(span); + } else { + spanEntry.setClientSpan(span); + } + } else { + spanEntry = new SpanEntry(); + if (span.isReceiver()) { + spanEntry.setServerSpan(span); + } else { + spanEntry.setClientSpan(span); + } + spanEntryMap.put(span.getParentLevel() + "." + span.getLevelId(), spanEntry); } } - - //入HBase库 - //找到首个Node的MD5 - //拼接MR任务的Key - context.write(new Text(""), chainInfo); - } - - private String convertViewPoint(Span span) { - return null; + return spanEntryMap; } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java index 4361cd43d..4e514eca8 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java @@ -12,6 +12,7 @@ public class ChainInfo implements Writable { private String chainToken; private ChainStatus chainStatus; private List nodes; + private String userId; public ChainInfo() { this.nodes = new ArrayList(); @@ -77,6 +78,14 @@ public class ChainInfo implements Writable { this.nodes = nodes; } + public String getUserId() { + return userId; + } + + public void setUserId(String userId) { + this.userId = userId; + } + public enum ChainStatus { NORMAL('N'), ABNORMAL('A'); private char value; 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 715e04e83..14074d2a4 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 @@ -5,15 +5,16 @@ public class ChainNode { private String nodeToken; private String viewPoint; - //TODO private String businessKey; - private long cost; private NodeStatus status; private String parentLevelId; private int levelId; private String callType; + // 不参与序列化 + private String userId; + public String getNodeToken() { return nodeToken; } @@ -70,6 +71,22 @@ public class ChainNode { return callType; } + public void setUserId(String userId) { + this.userId = userId; + } + + public String getUserId() { + return userId; + } + + public String getBusinessKey() { + return businessKey; + } + + public void setBusinessKey(String businessKey) { + this.businessKey = businessKey; + } + public enum NodeStatus { NORMAL('N'), ABNORMAL('A'); private char value; diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java index 835c3a2ae..55e54c871 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java @@ -14,7 +14,7 @@ public class CostMap { return costs.containsKey(parentLevel); } - public long get(String parentLevel) { + public Long get(String parentLevel) { return costs.get(parentLevel); } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java new file mode 100644 index 000000000..ea40cbafd --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/SpanEntry.java @@ -0,0 +1,157 @@ +package com.ai.cloud.skywalking.analysis.model; + +import com.ai.cloud.skywalking.protocol.CallType; +import com.ai.cloud.skywalking.protocol.Span; + +public class SpanEntry { + + private Span clientSpan; + private Span serverSpan; + + public SpanEntry() { + + } + + public int getLevelId() { + if (clientSpan != null) { + return clientSpan.getLevelId(); + } + + return serverSpan.getLevelId(); + } + + public String getParentLevelId() { + if (clientSpan != null) { + return clientSpan.getParentLevel(); + } + + return serverSpan.getParentLevel(); + } + + public String getViewPoint() { + if (clientSpan != null) { + return clientSpan.getViewPointId(); + } + + return serverSpan.getViewPointId(); + } + + public CallType getCallType() { + if (clientSpan != null) { + return CallType.convert(clientSpan.getCallType()); + } + + return CallType.convert(serverSpan.getCallType()); + } + + public long getCost() { + long resultCost = 0; + switch (getCallType()) { + case ASYNC: + resultCost = getClientCost() + getServerCost(); + break; + case SYNC: + resultCost = getClientCost(); + if (getClientSpan() == null) { + resultCost = getServerCost(); + } + break; + case LOCAL: + resultCost = getClientCost(); + break; + } + + return resultCost; + } + + public long getClientCost() { + if (clientSpan != null) { + return clientSpan.getCost(); + } + + return 0; + } + + public long getServerCost() { + if (serverSpan != null) { + return serverSpan.getCost(); + } + + return 0; + } + + public String getBusinessKey() { + if (clientSpan != null) { + return clientSpan.getBusinessKey(); + } + + return serverSpan.getBusinessKey(); + } + + public ChainNode.NodeStatus getSpanStatus() { + if (clientSpan != null) { + if (clientSpan.getExceptionStack() != null && clientSpan.getExceptionStack().length() > 0) { + return ChainNode.NodeStatus.ABNORMAL; + } + } + + if (serverSpan != null) { + if (serverSpan.getExceptionStack() != null && serverSpan.getExceptionStack().length() > 0) { + return ChainNode.NodeStatus.ABNORMAL; + } + } + + return ChainNode.NodeStatus.NORMAL; + } + + public String getExceptionStack() { + StringBuilder exceptionStack = new StringBuilder(); + if (clientSpan != null) { + exceptionStack.append("Client" + clientSpan.getExceptionStack()); + } + + if (serverSpan != null) { + exceptionStack.append("Server cause by :" + serverSpan.getExceptionStack()); + } + + return exceptionStack.toString(); + } + + public String getApplicationId() { + if (clientSpan != null) { + return clientSpan.getApplicationId(); + } + + return serverSpan.getApplicationId(); + } + + public Span getClientSpan() { + return clientSpan; + } + + public Span getServerSpan() { + return serverSpan; + } + + public void setClientSpan(Span clientSpan) { + this.clientSpan = clientSpan; + } + + public void setServerSpan(Span serverSpan) { + this.serverSpan = serverSpan; + } + + public String getSpanType() { + if (clientSpan != null) { + return clientSpan.getSpanType(); + } + return serverSpan.getSpanType(); + } + + public String getUserId() { + if (clientSpan != null) { + return clientSpan.getUserId(); + } + return serverSpan.getUserId(); + } +} 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 new file mode 100644 index 000000000..7c24e65d0 --- /dev/null +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/HBaseUtil.java @@ -0,0 +1,111 @@ +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; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.*; +import org.apache.hadoop.hbase.util.Bytes; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +public class HBaseUtil { + private static Logger logger = LoggerFactory.getLogger(HBaseUtil.class.getName()); + + private static Configuration configuration = null; + private static Connection connection; + + public static boolean saveData(String traceId, ChainInfo chainInfo) { + Table table = null; + + try { + table = connection.getTable(TableName.valueOf(Config.HBase.TRACE_INFO_TABLE_NAME)); + } catch (IOException e) { + logger.error("Cannot found table[" + Config.HBase.TRACE_INFO_TABLE_NAME + "]", e); + } + + 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())); + } + + try { + table.put(put); + if (logger.isDebugEnabled()) { + logger.debug("Insert data[RowKey:{}] success.", put.getId()); + } + } catch (IOException e) { + logger.error("Insert data [Rowkey:{}] failed.", put.getId(), e); + return false; + } + + return true; + } + + static { + try { + initHBaseClient(); + Admin admin = connection.getAdmin(); + if (!admin.isTableAvailable(TableName.valueOf(Config.HBase.TRACE_INFO_TABLE_NAME))) { + HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(Config.HBase.TRACE_INFO_TABLE_NAME)); + tableDesc.addFamily(new HColumnDescriptor(Config.HBase.TRACE_INFO_COLUMN_FAMILY)); + admin.createTable(tableDesc); + logger.info("Create table [{}] ok!", Config.HBase.TRACE_INFO_TABLE_NAME); + } + } catch (IOException e) { + logger.error("Create table[{}] failed", Config.HBase.TRACE_INFO_TABLE_NAME, e); + } + } + + private static void initHBaseClient() throws IOException { + if (configuration == null) { + configuration = HBaseConfiguration.create(); + if (Config.HBase.ZK_QUORUM == null || "".equals(Config.HBase.ZK_QUORUM)) { + logger.error("Miss HBase ZK quorum Configuration", new IllegalArgumentException("Miss HBase ZK quorum Configuration")); + System.exit(-1); + } + configuration.set("hbase.zookeeper.quorum", Config.HBase.ZK_QUORUM); + configuration.set("hbase.zookeeper.property.clientPort", Config.HBase.ZK_CLIENT_PORT); + connection = ConnectionFactory.createConnection(configuration); + } + } + +} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/ViewPointFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/ViewPointFilter.java deleted file mode 100644 index a81d9377d..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/ViewPointFilter.java +++ /dev/null @@ -1,18 +0,0 @@ -package com.ai.cloud.skywalking.analysis.viewpoint; - -import com.ai.cloud.skywalking.protocol.Span; - -public abstract class ViewPointFilter { - private ViewPointFilter viewPointFilter; - - public abstract void doFilter(Span viewPoint, String span); - - public ViewPointFilter getViewPointFilter() { - return viewPointFilter; - } - - public void setViewPointFilter(ViewPointFilter viewPointFilter) { - this.viewPointFilter = viewPointFilter; - } - -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/AppendBusinessKeyFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/AppendBusinessKeyFilter.java deleted file mode 100644 index 86da43c86..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/AppendBusinessKeyFilter.java +++ /dev/null @@ -1,15 +0,0 @@ -package com.ai.cloud.skywalking.analysis.viewpoint.impl; - -import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter; -import com.ai.cloud.skywalking.protocol.Span; - -public class AppendBusinessKeyFilter extends ViewPointFilter { - @Override - public void doFilter(Span span, String viewPoint) { - viewPoint += span.getBusinessKey(); - - if (getViewPointFilter() != null) { - getViewPointFilter().doFilter(span, viewPoint); - } - } -} diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/ReplaceIPPortFilter.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/ReplaceIPPortFilter.java deleted file mode 100644 index 5d03362a5..000000000 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/ReplaceIPPortFilter.java +++ /dev/null @@ -1,22 +0,0 @@ -package com.ai.cloud.skywalking.analysis.viewpoint.impl; - -import com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter; -import com.ai.cloud.skywalking.protocol.Span; - -public class ReplaceIPPortFilter extends ViewPointFilter { - - //ip:port regex - private static String IP_PORT_REGEX = "^([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\." + - "([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])" + - "|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5])):" + - "([0-9]{1,4}|[1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5])$"; - - @Override - public void doFilter(Span span, String viewPoint) { - viewPoint = viewPoint.replaceAll(IP_PORT_REGEX, span.getApplicationId()); - - if (getViewPointFilter() != null) { - getViewPointFilter().doFilter(span, viewPoint); - } - } -} diff --git a/skywalking-analysis/src/main/resources/config.properties b/skywalking-analysis/src/main/resources/config.properties index e69de29bb..57d720597 100644 --- a/skywalking-analysis/src/main/resources/config.properties +++ b/skywalking-analysis/src/main/resources/config.properties @@ -0,0 +1,15 @@ +#еİ +filter.filter_package_name=com.ai.cloud.skywalking.analysis.filter.impl +hbase.call_chain_table_name=sw-call-chain +hbase.zk_quorum=10.1.235.197,10.1.235.198,10.1.235.199 +hbase.zk_client_port=29181 +hbase.trace_info_table_name=sw-trace-info +hbase.trace_info_column_family=trace_info + +traceinfo.parent_level_id=parentLevelId +traceinfo.level_id=levelId +traceinfo.business_key=businessKey +traceinfo.cost=cost +traceinfo.trace_info_column_cid=cid +traceinfo.status=status +traceinfo.user_id=UId \ No newline at end of file diff --git a/skywalking-analysis/src/main/resources/viewpointfilter.conf b/skywalking-analysis/src/main/resources/viewpointfilter.conf index 813af5b76..788b82387 100644 --- a/skywalking-analysis/src/main/resources/viewpointfilter.conf +++ b/skywalking-analysis/src/main/resources/viewpointfilter.conf @@ -1,3 +1,3 @@ # -D,M,W=com.ai.cloud.skywalking.analysis.viewpoint.impl.ReplaceIPPortFilter -J=com.ai.cloud.skywalking.analysis.viewpoint.impl.AppendBusinessKeyFilter \ No newline at end of file +default=CopyAttrFilter->ReplaceAddressFilter->ProcessCostTimeFilter->TokenGenerateFilter +J=CopyAttrFilter->AppendBusinessKeyFilter->ProcessCostTimeFilter->TokenGenerateFilter 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 new file mode 100644 index 000000000..f9b190ce7 --- /dev/null +++ b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java @@ -0,0 +1,152 @@ +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; +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.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Created by astraea on 2016/1/15. + */ +public class CallChainMapperTest { + + private static String ZK_QUORUM = "10.1.235.197,10.1.235.198,10.1.235.199"; + private static String ZK_CLIENT_PORT = "29181"; + private static String chain_Id = "1.0a2.1452852040127.0664234.11036.55.1"; + + private static Configuration configuration = null; + private static Connection connection; + + @Test + 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 && span.isReceiver()) { + if (span.isReceiver()) { + spanEntry.setServerSpan(span); + } else { + spanEntry.setClientSpan(span); + } + } else { + spanEntry = new SpanEntry(); + if (span.isReceiver()) { + spanEntry.setServerSpan(span); + } else { + spanEntry.setClientSpan(span); + } + spanEntryMap.put(span.getParentLevel() + "." + span.getLevelId(), spanEntry); + } + } + return spanEntryMap; + } + + public static List selectByTraceId(String traceId) throws IOException { + List entries = new ArrayList(); + Table table = connection.getTable(TableName.valueOf(Config.HBase.CALL_CHAIN_TABLE_NAME)); + Get g = new Get(Bytes.toBytes(traceId)); + Result r = table.get(g); + for (Cell cell : r.rawCells()) { + if (cell.getValueArray().length > 0) + entries.add(new Span(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()))); + } + return entries; + } + + + @Before + public void initHBaseClient() throws IOException { + if (configuration == null) { + configuration = HBaseConfiguration.create(); + configuration.set("hbase.zookeeper.quorum", ZK_QUORUM); + configuration.set("hbase.zookeeper.property.clientPort", ZK_CLIENT_PORT); + connection = ConnectionFactory.createConnection(configuration); + } + } +} \ No newline at end of file