增加异常调用链分析

@@ -60,6 +60,8 @@ public class ApiStatisticsAnalyzer implements Serializable {
             result.setApiName(trace._2().getApi());
             result.setCallTimes(1);
             result.setTraceMd5(trace._1());
+            result.setErrorStatus(trace._2().isErrorStatus());
+            result.setTraceId(trace._2().getTraceid());
 
             List<SpanInfo> spanList = trace._2().getSortSpanList();
             List<SpanResult> spanResultList = new ArrayList<SpanResult>();
@@ -67,7 +69,7 @@ public class ApiStatisticsAnalyzer implements Serializable {
             while (itor.hasNext()) {
                 SpanInfo span = (SpanInfo) itor.next();
                 SpanResult spanResult = new SpanResult(span.getName(), span.getEnd() - span.getBegin(), span.getLevel(),
-                        span.getSpanid(), span.getParent(), span.getSrcService(), span.getDstService(), null, null);
+                        span.getSpanid(), span.getParent(), span.getSrcService(), span.getDstService(), null, null,span.isErrorStatus());
                 spanResultList.add(spanResult);
             }
 
@@ -77,6 +79,7 @@ public class ApiStatisticsAnalyzer implements Serializable {
             result.setMinLatencyTrace(trace._2().getTraceid());
             result.setMaxLatency((int) trace._2().getDuration());
             result.setMinLatency((int) trace._2().getDuration());
+            result.setTraceStartTime(trace._2().getTraceStartTime());
 
             return new Tuple2<String, ApiTraceResult>(trace._1(), result);
         }
@@ -173,8 +176,14 @@ public class ApiStatisticsAnalyzer implements Serializable {
         String api = rootSpan.getName() ;
         StringBuilder key = new StringBuilder() ;
         Iterator it = sortSpanList.iterator() ;
+        //判断该treace是否是异常调用链
+        boolean traceErrorStatus = false;
         while(it.hasNext()) {
-            key.append( ((SpanInfo)it.next()).getName()+"|");
+            SpanInfo si = (SpanInfo)it.next();
+            key.append(si.getName()+"|");
+            if(si.isErrorStatus()){
+                traceErrorStatus = true;
+            }
         }
 
         String keyMd5 = MD5.md5(key.toString());
@@ -184,7 +193,9 @@ public class ApiStatisticsAnalyzer implements Serializable {
         trace.setDuration(duration);
         trace.setTraceid(rootSpan.getTraceid());
         trace.setSortSpanList(sortSpanList);
-
+        trace.setErrorStatus(traceErrorStatus);
+        //root开始时间当做trace的开始时间
+        trace.setTraceStartTime(rootSpan.getBegin());
 
         //毫秒转换为 秒，丢失 毫秒数
         long beginSec = rootSpan.getReceive()/1000  ;

@@ -29,4 +29,10 @@ public class ApiTraceResult implements Serializable {
 
     String prefex ;
 
+    String traceId;
+
+    boolean errorStatus;
+
+    long traceStartTime;
+
 }
\ No newline at end of file

@@ -19,4 +19,7 @@ public class SortedTrace implements Serializable {
     long startMinute ;
     long startDay ;
 
+    boolean errorStatus;
+    long traceStartTime;
+
 }
\ No newline at end of file

@@ -33,4 +33,7 @@ public class SpanInfo implements Serializable {
     private String pageId;
 
     private String httpHost;
+
+    private boolean errorStatus;
+    long duration;
 }
\ No newline at end of file

@@ -23,9 +23,10 @@ public class SpanResult implements Serializable {
 
     List<String> srcIp ;
     List<String> dstIp ;
+    boolean errorStatus;
 
 
-    public SpanResult(String spanName, long duration,int level,String spanId, String parent, String srcService, String dstService, List<String> srcIp, List<String> dstIp  ){
+    public SpanResult(String spanName, long duration,int level,String spanId, String parent, String srcService, String dstService, List<String> srcIp, List<String> dstIp,boolean errorStatus  ){
         this.spanName = spanName ;
         this.spanId = spanId ;
         this.duration = duration ;
@@ -36,6 +37,7 @@ public class SpanResult implements Serializable {
         this.dstService = dstService ;
         this.srcIp = srcIp ;
         this.dstIp = dstIp ;
+        this.errorStatus = errorStatus;
     }
 
 }

@@ -116,7 +116,7 @@ public class ApiAnalyzeHandler implements IAnalyzeHandler, Serializable {
         List<SpanResult> list = new ArrayList();
         for (int i = 0; i < spanList.size(); i++) {
             list.add(new SpanResult(spanList.get(i).getName(), durationPerStep.get(String.valueOf(spanList.get(i).getName())),
-                    spanList.get(i).getLevel(), spanList.get(i).getSpanid() ,spanList.get(i).getParent() , spanList.get(i).getSrcService(), spanList.get(i).getDstService(), null, null));
+                    spanList.get(i).getLevel(), spanList.get(i).getSpanid() ,spanList.get(i).getParent() , spanList.get(i).getSrcService(), spanList.get(i).getDstService(), null, null,spanList.get(i).isErrorStatus()));
         }
 
 

@@ -79,6 +79,7 @@ public class TraceAnalyzeHandler implements TraceHandler, Serializable {
                         spanInfo.setName(span.getName());
                         spanInfo.setBegin(span.getBegin());
                         spanInfo.setEnd(span.getEnd());
+                        spanInfo.setDuration(span.getEnd() - span.getBegin());
                         spanInfo.setTraceid(Span.idToHex(span.getTraceId()));
                         spanInfo.setSpanid(Span.idToHex(span.getSpanId()));
                         if (span.getParents().size() > 0) {
@@ -90,7 +91,12 @@ public class TraceAnalyzeHandler implements TraceHandler, Serializable {
                         spanInfo.setReceive(spans.getReceive());
                         if(span.tags()!=null){
                             spanInfo.setHttpHost(span.tags().get("http.host"));
+                            //标记span是否是是异常span
+                            if(StringUtils.isNotBlank(span.tags().get("error"))){
+                                spanInfo.setErrorStatus(true);
+                            }
                         }
+
                         spanInfoList.add(new Tuple2<>(spanInfo.getTraceid(), spanInfo));
                     }
                 }
@@ -247,23 +253,17 @@ public class TraceAnalyzeHandler implements TraceHandler, Serializable {
 
         sortSpanTrace.cache();
 
-        //处理span+ip的耗时分布
+        //1处理span+ip的耗时分布
         handlerSpanIp(sortSpanTrace);
 
         JavaPairDStream<String, ApiTraceResult> apiResultTraceDStream = sortSpanTrace.mapToPair(analyzer.ConvertTraceResultFunc) ;
+        apiResultTraceDStream.cache();
 
-        //根据traceMD5分组计算value
-        JavaPairDStream<String, ApiTraceResult> resultDStream = apiResultTraceDStream.reduceByKey(analyzer.ReduceFunc) ;
-
-        resultDStream.foreachRDD(new VoidFunction<JavaPairRDD<String, ApiTraceResult>>() {
-            @Override
-            public void call(JavaPairRDD<String, ApiTraceResult> apiResultRdd) throws Exception {
-
-                ApiStatisticsResultStore.store(apiResultRdd, "trace_api_analyze_minutes");
-            }
-        });
-
+        //2处理treemd5调用链总览
+        handlertraceApiAnalyzeMinutes(apiResultTraceDStream);
 
+        //3处理异常调用链信息
+        handlerExceptionTrace(apiResultTraceDStream);
     }
 
 
@@ -369,4 +369,78 @@ public class TraceAnalyzeHandler implements TraceHandler, Serializable {
         });
 
     }
+
+
+
+    private void handlertraceApiAnalyzeMinutes(JavaPairDStream<String, ApiTraceResult> apiResultTraceDStream){
+        //根据traceMD5分组计算value
+        JavaPairDStream<String, ApiTraceResult> resultDStream = apiResultTraceDStream.reduceByKey(analyzer.ReduceFunc) ;
+
+        resultDStream.foreachRDD(new VoidFunction<JavaPairRDD<String, ApiTraceResult>>() {
+            @Override
+            public void call(JavaPairRDD<String, ApiTraceResult> apiResultRdd) throws Exception {
+
+                ApiStatisticsResultStore.store(apiResultRdd, "trace_api_analyze_minutes");
+            }
+        });
+    }
+
+    private void handlerExceptionTrace(JavaPairDStream<String, ApiTraceResult> apiResultTraceDStream){
+        JavaPairDStream<String, ApiTraceResult> filter = apiResultTraceDStream.filter(new Function<Tuple2<String, ApiTraceResult>, Boolean>() {
+            @Override
+            public Boolean call(Tuple2<String, ApiTraceResult> stringSortedTraceTuple2) throws Exception {
+                if(stringSortedTraceTuple2._2.isErrorStatus()){
+                    return true;
+                }else{
+                    return false;
+                }
+            }
+        });
+
+
+
+
+        filter.foreachRDD(new VoidFunction<JavaPairRDD<String, ApiTraceResult>>() {
+            @Override
+            public void call(JavaPairRDD<String, ApiTraceResult> stringSortedTraceJavaPairRDD) throws Exception {
+                stringSortedTraceJavaPairRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, ApiTraceResult>>>() {
+                    @Override
+                    public void call(Iterator<Tuple2<String, ApiTraceResult>> tuple2Iterator) throws Exception {
+                        HTable resultTable = null;
+                        try {
+                            if(tuple2Iterator == null){
+                                return;
+                            }
+
+                            if (resultTable == null) {
+                                resultTable = (HTable) HBasePool.getConnection().getTable(TableName.valueOf("exception_trace"));
+                            }
+
+                            List<Put> putList = new ArrayList<>();
+                            while(tuple2Iterator.hasNext()){
+                                Tuple2<String, ApiTraceResult> next = tuple2Iterator.next();
+                                ApiTraceResult apiTraceResult = next._2;
+
+                                Put put = new Put(Bytes.toBytes(apiTraceResult.getTraceStartTime()/1000 + ":" + apiTraceResult.getTraceId()));
+                                put.addColumn(Bytes.toBytes("data"), Bytes.toBytes("spans"), Bytes.toBytes(JSONObject.toJSONString(apiTraceResult.getSpans())));
+                                put.addColumn(Bytes.toBytes("data"), Bytes.toBytes("traceid"), Bytes.toBytes(apiTraceResult.getTraceId()));
+                                put.addColumn(Bytes.toBytes("data"), Bytes.toBytes("starttime"), Bytes.toBytes(apiTraceResult.getTraceStartTime()/1000));
+
+                                putList.add(put);
+                            }
+
+                            resultTable.put(putList);
+
+
+                        } catch (Exception e) {
+                            logger.error(e.getMessage(),e);
+                        } finally {
+                            if (resultTable != null)
+                                resultTable.close();
+                        }
+                    }
+                });
+            }
+        });
+    }
 }