mock mock mock

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/RecordPersistenceMember.java

@@ -22,18 +22,24 @@ public abstract class RecordPersistenceMember extends PersistenceMember {
 
     private Logger logger = LogManager.getFormatterLogger(RecordPersistenceMember.class);
 
-    private RecordPersistenceData persistenceData = new RecordPersistenceData();
+    private RecordPersistenceData persistenceData;
 
     public RecordPersistenceMember(Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
         super(role, clusterContext, selfContext);
+        persistenceData = new RecordPersistenceData();
+    }
+
+    private RecordPersistenceData getPersistenceData() {
+        return this.persistenceData;
     }
 
     @Override
     public void analyse(Object message) throws Exception {
         if (message instanceof RecordData) {
             RecordData recordData = (RecordData) message;
-            persistenceData.getElseCreate(recordData.getId()).setRecord(recordData.getRecord());
-            if (persistenceData.size() >= WorkerConfig.Persistence.Data.size) {
+            logger.debug("setRecord: id: %s, data: %s", recordData.getId(), recordData.getRecord());
+            getPersistenceData().getElseCreate(recordData.getId()).setRecord(recordData.getRecord());
+            if (getPersistenceData().size() >= WorkerConfig.Persistence.Data.size) {
                 persistence();
             }
         } else {
@@ -44,19 +50,20 @@ public abstract class RecordPersistenceMember extends PersistenceMember {
     protected void persistence() {
         boolean success = saveToEs();
         if (success) {
-            persistenceData.clear();
+            getPersistenceData().clear();
         }
     }
 
     private boolean saveToEs() {
         Client client = EsClient.getClient();
         BulkRequestBuilder bulkRequest = client.prepareBulk();
-        logger.debug("persistenceData size: %s", persistenceData.size());
+        logger.debug("persistenceData size: %s", getPersistenceData().size());
 
-        Iterator<Map.Entry<String, RecordData>> iterator = persistenceData.iterator();
+        Iterator<Map.Entry<String, RecordData>> iterator = getPersistenceData().iterator();
 
         while (iterator.hasNext()) {
             Map.Entry<String, RecordData> recordData = iterator.next();
+            logger.debug("saveToEs: key: %s, data: %s", recordData.getKey(), recordData.getValue().getRecord().toString());
             bulkRequest.add(client.prepareIndex(esIndex(), esType(), recordData.getKey()).setSource(recordData.getValue().getRecord().toString()));
         }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/WorkerConfig.java

@@ -46,6 +46,43 @@ public class WorkerConfig extends ClusterConfig {
     }
 
     public static class Queue {
+        public static class Segment {
+            public static class SegmentCostSave {
+                public static int Size = 1024;
+            }
+
+            public static class SegmentSave {
+                public static int Size = 1024;
+            }
+
+            public static class SegmentExceptionSave {
+                public static int Size = 1024;
+            }
+        }
+
+        public static class Node {
+            public static class NodeDayAnalysis {
+                public static int Size = 1024;
+            }
+
+            public static class NodeRefDayAnalysis {
+                public static int Size = 1024;
+            }
+
+            public static class NodeHourAnalysis {
+                public static int Size = 1024;
+            }
+
+            public static class NodeMinuteAnalysis {
+                public static int Size = 1024;
+            }
+
+            public static class NodeRefMinuteAnalysis {
+                public static int Size = 1024;
+            }
+        }
+
+
         public static class Persistence {
             public static class DAGNodePersistence {
                 public static int Size = 1024;

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/node/NodeIndex.java

@@ -16,6 +16,7 @@ public class NodeIndex extends AbstractIndex {
     public static final String Code = "code";
     public static final String NickName = "nickName";
     public static final String Component = "component";
+    public static final String IsReal = "isReal";
 
     @Override
     public String index() {
@@ -44,6 +45,10 @@ public class NodeIndex extends AbstractIndex {
                             .field("type", "string")
                             .field("index", "not_analyzed")
                         .endObject()
+                        .startObject(IsReal)
+                            .field("type", "boolean")
+                            .field("index", "not_analyzed")
+                        .endObject()
                         .startObject(AGG_COLUMN)
                             .field("type", "string")
                             .field("index", "not_analyzed")

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/node/analysis/AbstractNodeAnalysis.java

@@ -39,6 +39,7 @@ abstract class AbstractNodeAnalysis extends RecordAnalysisMember {
             JsonObject appDataJsonObj = new JsonObject();
             appDataJsonObj.addProperty(NodeIndex.Code, segment.getApplicationCode());
             appDataJsonObj.addProperty(NodeIndex.NickName, segment.getApplicationCode());
+            appDataJsonObj.addProperty(NodeIndex.IsReal, true);
             appDataJsonObj.addProperty(NodeIndex.Time_Slice, timeSlice);
             setRecord(id, appDataJsonObj);
 
@@ -47,6 +48,7 @@ abstract class AbstractNodeAnalysis extends RecordAnalysisMember {
                 String kind = Tags.SPAN_KIND.get(span);
                 String component = Tags.COMPONENT.get(span);
                 dataJsonObj.addProperty(NodeIndex.Component, component);
+                dataJsonObj.addProperty(NodeIndex.IsReal, true);
 
                 String code = segment.getApplicationCode();
                 dataJsonObj.addProperty(NodeIndex.Code, code);
@@ -59,6 +61,7 @@ abstract class AbstractNodeAnalysis extends RecordAnalysisMember {
                     code = SpanPeersTools.getPeers(span);
                     dataJsonObj.addProperty(NodeIndex.Code, code);
                     dataJsonObj.addProperty(NodeIndex.NickName, code);
+                    dataJsonObj.addProperty(NodeIndex.IsReal, false);
 
                     id = timeSlice + Const.ID_SPLIT + code;
                     logger.debug("leaf client node: %s", dataJsonObj.toString());
@@ -71,6 +74,7 @@ abstract class AbstractNodeAnalysis extends RecordAnalysisMember {
                         userDataJsonObj.addProperty(NodeIndex.Component, Const.USER_CODE);
                         userDataJsonObj.addProperty(NodeIndex.NickName, Const.USER_CODE);
                         userDataJsonObj.addProperty(NodeIndex.Time_Slice, timeSlice);
+                        userDataJsonObj.addProperty(NodeIndex.IsReal, false);
                         String userId = timeSlice + Const.ID_SPLIT + Const.USER_CODE;
                         logger.debug("user node: %s", userDataJsonObj.toString());
                         setRecord(userId, userDataJsonObj);
@@ -83,7 +87,7 @@ abstract class AbstractNodeAnalysis extends RecordAnalysisMember {
                         for (TraceSegmentRef segmentRef : segment.getRefs()) {
                             String nickName = Const.PEERS_FRONT_SPLIT + segmentRef.getPeerHost() + Const.PEERS_BEHIND_SPLIT;
                             dataJsonObj.addProperty(NodeIndex.NickName, nickName);
-                            id = timeSlice + Const.ID_SPLIT + code;
+                            id = timeSlice + Const.ID_SPLIT + nickName;
                             logger.debug("refs node: %s", dataJsonObj.toString());
                             setRecord(id, dataJsonObj);
                         }

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/node/analysis/NodeDayAnalysis.java

@@ -5,6 +5,7 @@ import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
 import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
 import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
 import com.a.eye.skywalking.collector.actor.selector.WorkerSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
 import com.a.eye.skywalking.collector.worker.node.persistence.NodeDayAgg;
 import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
 import com.a.eye.skywalking.collector.worker.storage.RecordData;
@@ -51,7 +52,7 @@ public class NodeDayAnalysis extends AbstractNodeAnalysis {
 
         @Override
         public int queueSize() {
-            return 1024;
+            return WorkerConfig.Queue.Node.NodeDayAnalysis.Size;
         }
     }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/node/analysis/NodeHourAnalysis.java

@@ -5,6 +5,7 @@ import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
 import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
 import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
 import com.a.eye.skywalking.collector.actor.selector.WorkerSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
 import com.a.eye.skywalking.collector.worker.node.persistence.NodeHourAgg;
 import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
 import com.a.eye.skywalking.collector.worker.storage.RecordData;
@@ -51,7 +52,7 @@ public class NodeHourAnalysis extends AbstractNodeAnalysis {
 
         @Override
         public int queueSize() {
-            return 1024;
+            return WorkerConfig.Queue.Node.NodeHourAnalysis.Size;
         }
     }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/node/analysis/NodeMinuteAnalysis.java

@@ -5,6 +5,7 @@ import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
 import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
 import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
 import com.a.eye.skywalking.collector.actor.selector.WorkerSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
 import com.a.eye.skywalking.collector.worker.node.persistence.NodeMinuteAgg;
 import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
 import com.a.eye.skywalking.collector.worker.storage.RecordData;
@@ -51,7 +52,7 @@ public class NodeMinuteAnalysis extends AbstractNodeAnalysis {
 
         @Override
         public int queueSize() {
-            return 1024;
+            return WorkerConfig.Queue.Node.NodeMinuteAnalysis.Size;
         }
     }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/noderef/analysis/NodeRefDayAnalysis.java

@@ -17,7 +17,7 @@ import com.a.eye.skywalking.trace.TraceSegment;
  */
 public class NodeRefDayAnalysis extends AbstractNodeRefAnalysis {
 
-    private NodeRefDayAnalysis(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
+    protected NodeRefDayAnalysis(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
         super(role, clusterContext, selfContext);
     }
 
@@ -70,7 +70,7 @@ public class NodeRefDayAnalysis extends AbstractNodeRefAnalysis {
 
         @Override
         public int queueSize() {
-            return WorkerConfig.Queue.DAGNodeRefAnalysis.Size;
+            return WorkerConfig.Queue.Node.NodeRefDayAnalysis.Size;
         }
     }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/noderef/analysis/NodeRefMinuteAnalysis.java

@@ -17,7 +17,7 @@ import com.a.eye.skywalking.trace.TraceSegment;
  */
 public class NodeRefMinuteAnalysis extends AbstractNodeRefAnalysis {
 
-    private NodeRefMinuteAnalysis(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
+    protected NodeRefMinuteAnalysis(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
         super(role, clusterContext, selfContext);
     }
 
@@ -70,7 +70,7 @@ public class NodeRefMinuteAnalysis extends AbstractNodeRefAnalysis {
 
         @Override
         public int queueSize() {
-            return WorkerConfig.Queue.DAGNodeRefAnalysis.Size;
+            return WorkerConfig.Queue.Node.NodeRefMinuteAnalysis.Size;
         }
     }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/segment/persistence/SegmentCostSave.java

@@ -10,7 +10,6 @@ import com.a.eye.skywalking.collector.worker.RecordPersistenceMember;
 import com.a.eye.skywalking.collector.worker.WorkerConfig;
 import com.a.eye.skywalking.collector.worker.segment.SegmentCostIndex;
 import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
-import com.a.eye.skywalking.collector.worker.storage.AbstractIndex;
 import com.a.eye.skywalking.collector.worker.storage.RecordData;
 import com.a.eye.skywalking.collector.worker.tools.CollectionTools;
 import com.a.eye.skywalking.trace.Span;
@@ -33,10 +32,10 @@ public class SegmentCostSave extends RecordPersistenceMember {
 
     @Override
     public String esType() {
-        return AbstractIndex.Type_Record;
+        return SegmentCostIndex.Type_Record;
     }
 
-    private SegmentCostSave(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
+    protected SegmentCostSave(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
         super(role, clusterContext, selfContext);
     }
 
@@ -85,7 +84,7 @@ public class SegmentCostSave extends RecordPersistenceMember {
 
         @Override
         public int queueSize() {
-            return WorkerConfig.Queue.TraceSegmentRecordAnalysis.Size;
+            return WorkerConfig.Queue.Segment.SegmentCostSave.Size;
         }
 
         @Override

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/segment/persistence/SegmentExceptionSave.java

@@ -41,7 +41,7 @@ public class SegmentExceptionSave extends RecordPersistenceMember {
         return AbstractIndex.Type_Record;
     }
 
-    private SegmentExceptionSave(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
+    protected SegmentExceptionSave(com.a.eye.skywalking.collector.actor.Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
         super(role, clusterContext, selfContext);
     }
 
@@ -90,7 +90,7 @@ public class SegmentExceptionSave extends RecordPersistenceMember {
 
         @Override
         public int queueSize() {
-            return WorkerConfig.Queue.TraceSegmentRecordAnalysis.Size;
+            return WorkerConfig.Queue.Segment.SegmentExceptionSave.Size;
         }
 
         @Override

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/segment/persistence/SegmentSave.java

@@ -58,7 +58,7 @@ public class SegmentSave extends RecordPersistenceMember {
 
         @Override
         public int queueSize() {
-            return WorkerConfig.Queue.TraceSegmentRecordAnalysis.Size;
+            return WorkerConfig.Queue.Segment.SegmentSave.Size;
         }
 
         @Override

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/mock/MockEsBulkClient.java

+package com.a.eye.skywalking.collector.worker.mock;
+
+import com.a.eye.skywalking.collector.worker.storage.EsClient;
+import org.elasticsearch.action.ListenableActionFuture;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Client;
+import org.mockito.Mockito;
+
+import static org.powermock.api.mockito.PowerMockito.*;
+
+/**
+ * @author pengys5
+ */
+public class MockEsBulkClient {
+
+    public IndexRequestBuilder indexRequestBuilder;
+
+    public void createMock() {
+        Client client = mock(Client.class);
+        mockStatic(EsClient.class);
+        when(EsClient.getClient()).thenReturn(client);
+
+        BulkRequestBuilder bulkRequestBuilder = mock(BulkRequestBuilder.class);
+        when(client.prepareBulk()).thenReturn(bulkRequestBuilder);
+
+        ListenableActionFuture listenableActionFuture = mock(ListenableActionFuture.class);
+        when(bulkRequestBuilder.execute()).thenReturn(listenableActionFuture);
+
+        BulkResponse responses = mock(BulkResponse.class);
+        when(listenableActionFuture.actionGet()).thenReturn(responses);
+
+        when(responses.hasFailures()).thenReturn(true);
+
+        indexRequestBuilder = mock(IndexRequestBuilder.class);
+
+        when(client.prepareIndex(Mockito.anyString(), Mockito.anyString(), Mockito.anyString())).thenReturn(indexRequestBuilder);
+
+        when(indexRequestBuilder.setSource(Mockito.anyString())).thenReturn(indexRequestBuilder);
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/mock/RecordDataAnswer.java

+package com.a.eye.skywalking.collector.worker.mock;
+
+import com.a.eye.skywalking.collector.worker.storage.RecordData;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public class RecordDataAnswer implements Answer<Object> {
+
+    private Logger logger = LogManager.getFormatterLogger(RecordDataAnswer.class);
+
+    public RecordObj recordObj = new RecordObj();
+
+    @Override
+    public Object answer(InvocationOnMock invocation) throws Throwable {
+        RecordData recordData = (RecordData) invocation.getArguments()[0];
+        recordObj.setSource(recordData);
+        return null;
+    }
+
+    public class RecordObj {
+        private Map<String, RecordData> recordDataMap;
+
+        public RecordObj() {
+            recordDataMap = new HashMap<>();
+        }
+
+        public Map<String, RecordData> getRecordDataMap() {
+            return recordDataMap;
+        }
+
+        public void setSource(RecordData recordData) {
+            this.recordDataMap.put(recordData.getId(), recordData);
+            logger.info("id: %s, data: %s", recordData.getId(), recordData.getRecord().toString());
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/mock/SaveToEsSourceAnswer.java

+package com.a.eye.skywalking.collector.worker.mock;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/**
+ * @author pengys5
+ */
+public class SaveToEsSourceAnswer implements Answer<Object> {
+
+    private Logger logger = LogManager.getFormatterLogger(SaveToEsSourceAnswer.class);
+
+    public SourceObj sourceObj = new SourceObj();
+
+    @Override
+    public Object answer(InvocationOnMock invocation) throws Throwable {
+        Gson gson = new Gson();
+        String source = (String) invocation.getArguments()[0];
+        JsonObject sourceJsonObj = gson.fromJson(source, JsonObject.class);
+        logger.debug("sourceJsonObj: %s", sourceJsonObj.toString());
+
+        sourceObj.setSource(sourceJsonObj);
+        return null;
+    }
+
+    public class SourceObj {
+        private JsonArray source;
+
+        public SourceObj() {
+            source = new JsonArray();
+        }
+
+        public JsonArray getSource() {
+            return source;
+        }
+
+        public void setSource(JsonObject source) {
+            this.source.add(source);
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/node/analysis/NodeAnalysisVerfiy.java

+package com.a.eye.skywalking.collector.worker.node.analysis;
+
+import com.a.eye.skywalking.collector.worker.storage.RecordData;
+import org.junit.Assert;
+
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public enum NodeAnalysisVerfiy {
+    INSTANCE;
+
+    public void verfiyCacheService(Map<String, RecordData> recordDataMap, long timeSlice) {
+        RecordData data_1 = recordDataMap.get(timeSlice + "..-..[localhost:-1]");
+        Assert.assertEquals("H2", data_1.getRecord().get("component").getAsString());
+        Assert.assertEquals(false, data_1.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("[localhost:-1]", data_1.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_1.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("[localhost:-1]", data_1.getRecord().get("nickName").getAsString());
+
+        RecordData data_2 = recordDataMap.get(timeSlice + "..-..[127.0.0.1:6379]");
+        Assert.assertEquals("Redis", data_2.getRecord().get("component").getAsString());
+        Assert.assertEquals(false, data_2.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("[127.0.0.1:6379]", data_2.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_2.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("[127.0.0.1:6379]", data_2.getRecord().get("nickName").getAsString());
+
+        RecordData data_3 = recordDataMap.get(timeSlice + "..-..cache-service");
+        Assert.assertEquals("Motan", data_3.getRecord().get("component").getAsString());
+        Assert.assertEquals(true, data_3.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("cache-service", data_3.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_3.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("[127.0.0.1:8002]", data_3.getRecord().get("nickName").getAsString());
+    }
+
+    public void verfiyPortalService(Map<String, RecordData> recordDataMap, long timeSlice) {
+        RecordData data_1 = recordDataMap.get(timeSlice + "..-..portal-service");
+        Assert.assertEquals("Tomcat", data_1.getRecord().get("component").getAsString());
+        Assert.assertEquals(true, data_1.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("portal-service", data_1.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_1.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("portal-service", data_1.getRecord().get("nickName").getAsString());
+
+        RecordData data_2 = recordDataMap.get(timeSlice + "..-..[10.128.35.80:20880]");
+        Assert.assertEquals("HttpClient", data_2.getRecord().get("component").getAsString());
+        Assert.assertEquals(false, data_2.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("[10.128.35.80:20880]", data_2.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_2.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("[10.128.35.80:20880]", data_2.getRecord().get("nickName").getAsString());
+
+        RecordData data_3 = recordDataMap.get(timeSlice + "..-..[127.0.0.1:8002]");
+        Assert.assertEquals("Motan", data_3.getRecord().get("component").getAsString());
+        Assert.assertEquals(false, data_3.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("[127.0.0.1:8002]", data_3.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_3.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("[127.0.0.1:8002]", data_3.getRecord().get("nickName").getAsString());
+
+        RecordData data_4 = recordDataMap.get(timeSlice + "..-..User");
+        Assert.assertEquals("User", data_4.getRecord().get("component").getAsString());
+        Assert.assertEquals(false, data_4.getRecord().get("isReal").getAsBoolean());
+    }
+
+    public void verfiyPersistenceService(Map<String, RecordData> recordDataMap, long timeSlice) {
+        RecordData data_1 = recordDataMap.get(timeSlice + "..-..persistence-service");
+        Assert.assertEquals("Tomcat", data_1.getRecord().get("component").getAsString());
+        Assert.assertEquals(true, data_1.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("persistence-service", data_1.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_1.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("[10.128.35.80:20880]", data_1.getRecord().get("nickName").getAsString());
+
+        RecordData data_2 = recordDataMap.get(timeSlice + "..-..[127.0.0.1:3307]");
+        Assert.assertEquals("Mysql", data_2.getRecord().get("component").getAsString());
+        Assert.assertEquals(false, data_2.getRecord().get("isReal").getAsBoolean());
+        Assert.assertEquals("[127.0.0.1:3307]", data_2.getRecord().get("code").getAsString());
+        Assert.assertEquals(timeSlice, data_2.getRecord().get("timeSlice").getAsLong());
+        Assert.assertEquals("[127.0.0.1:3307]", data_2.getRecord().get("nickName").getAsString());
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/node/analysis/NodeDayAnalysisTestCase.java

+package com.a.eye.skywalking.collector.worker.node.analysis;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.WorkerRefs;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.RecordDataAnswer;
+import com.a.eye.skywalking.collector.worker.node.persistence.NodeDayAgg;
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.segment.mock.SegmentMock;
+import com.a.eye.skywalking.collector.worker.storage.RecordData;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.List;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClusterWorkerContext.class})
+@PowerMockIgnore({"javax.management.*"})
+public class NodeDayAnalysisTestCase {
+
+    private NodeDayAnalysis nodeDayAnalysis;
+    private SegmentMock segmentMock = new SegmentMock();
+    private RecordDataAnswer recordDataAnswer;
+
+    @Before
+    public void init() throws Exception {
+        ClusterWorkerContext clusterWorkerContext = PowerMockito.mock(ClusterWorkerContext.class);
+        WorkerRefs workerRefs = mock(WorkerRefs.class);
+        recordDataAnswer = new RecordDataAnswer();
+        doAnswer(recordDataAnswer).when(workerRefs).tell(Mockito.any(RecordData.class));
+
+        when(clusterWorkerContext.lookup(NodeDayAgg.Role.INSTANCE)).thenReturn(workerRefs);
+
+        LocalWorkerContext localWorkerContext = new LocalWorkerContext();
+        nodeDayAnalysis = new NodeDayAnalysis(NodeDayAnalysis.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(NodeDayAnalysis.class.getSimpleName(), NodeDayAnalysis.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), NodeDayAnalysis.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(NodeDayAnalysis.class.getSimpleName(), NodeDayAnalysis.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(NodeDayAnalysis.class.getSimpleName(), NodeDayAnalysis.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Node.NodeDayAnalysis.Size = testSize;
+        Assert.assertEquals(testSize, NodeDayAnalysis.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testCacheServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockCacheServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeDayAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyCacheService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310000L);
+    }
+
+    @Test
+    public void testPortalServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPortalServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeDayAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyPortalService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310000L);
+    }
+
+    @Test
+    public void testPersistenceServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPersistenceServiceSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeDayAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyPersistenceService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310000L);
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/node/analysis/NodeHourAnalysisTestCase.java

+package com.a.eye.skywalking.collector.worker.node.analysis;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.WorkerRefs;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.RecordDataAnswer;
+import com.a.eye.skywalking.collector.worker.node.persistence.NodeHourAgg;
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.segment.mock.SegmentMock;
+import com.a.eye.skywalking.collector.worker.storage.RecordData;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.List;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClusterWorkerContext.class})
+@PowerMockIgnore({"javax.management.*"})
+public class NodeHourAnalysisTestCase {
+
+    private NodeHourAnalysis nodeHourAnalysis;
+    private SegmentMock segmentMock = new SegmentMock();
+    private RecordDataAnswer recordDataAnswer;
+
+    @Before
+    public void init() throws Exception {
+        ClusterWorkerContext clusterWorkerContext = PowerMockito.mock(ClusterWorkerContext.class);
+        WorkerRefs workerRefs = mock(WorkerRefs.class);
+        recordDataAnswer = new RecordDataAnswer();
+        doAnswer(recordDataAnswer).when(workerRefs).tell(Mockito.any(RecordData.class));
+
+        when(clusterWorkerContext.lookup(NodeHourAgg.Role.INSTANCE)).thenReturn(workerRefs);
+
+        LocalWorkerContext localWorkerContext = new LocalWorkerContext();
+        nodeHourAnalysis = new NodeHourAnalysis(NodeHourAnalysis.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(NodeHourAnalysis.class.getSimpleName(), NodeHourAnalysis.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), NodeHourAnalysis.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(NodeHourAnalysis.class.getSimpleName(), NodeHourAnalysis.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(NodeHourAnalysis.class.getSimpleName(), NodeHourAnalysis.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Node.NodeHourAnalysis.Size = testSize;
+        Assert.assertEquals(testSize, NodeHourAnalysis.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testCacheServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockCacheServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeHourAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyCacheService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310900L);
+    }
+
+    @Test
+    public void testPortalServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPortalServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeHourAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyPortalService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310900L);
+    }
+
+    @Test
+    public void testPersistenceServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPersistenceServiceSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeHourAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyPersistenceService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310900L);
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/node/analysis/NodeMinuteAnalysisTestCase.java

+package com.a.eye.skywalking.collector.worker.node.analysis;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.WorkerRefs;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.RecordDataAnswer;
+import com.a.eye.skywalking.collector.worker.node.persistence.NodeMinuteAgg;
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.segment.mock.SegmentMock;
+import com.a.eye.skywalking.collector.worker.storage.RecordData;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.List;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClusterWorkerContext.class})
+@PowerMockIgnore({"javax.management.*"})
+public class NodeMinuteAnalysisTestCase {
+
+    private NodeMinuteAnalysis nodeMinuteAnalysis;
+    private SegmentMock segmentMock = new SegmentMock();
+    private RecordDataAnswer recordDataAnswer;
+
+    @Before
+    public void init() throws Exception {
+        ClusterWorkerContext clusterWorkerContext = PowerMockito.mock(ClusterWorkerContext.class);
+        WorkerRefs workerRefs = mock(WorkerRefs.class);
+        recordDataAnswer = new RecordDataAnswer();
+        doAnswer(recordDataAnswer).when(workerRefs).tell(Mockito.any(RecordData.class));
+
+        when(clusterWorkerContext.lookup(NodeMinuteAgg.Role.INSTANCE)).thenReturn(workerRefs);
+
+        LocalWorkerContext localWorkerContext = new LocalWorkerContext();
+        nodeMinuteAnalysis = new NodeMinuteAnalysis(NodeMinuteAnalysis.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(NodeMinuteAnalysis.class.getSimpleName(), NodeMinuteAnalysis.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), NodeMinuteAnalysis.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(NodeMinuteAnalysis.class.getSimpleName(), NodeMinuteAnalysis.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(NodeMinuteAnalysis.class.getSimpleName(), NodeMinuteAnalysis.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Node.NodeMinuteAnalysis.Size = testSize;
+        Assert.assertEquals(testSize, NodeMinuteAnalysis.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testCacheServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockCacheServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeMinuteAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyCacheService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310915L);
+    }
+
+    @Test
+    public void testPortalServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPortalServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeMinuteAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyPortalService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310915L);
+    }
+
+    @Test
+    public void testPersistenceServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPersistenceServiceSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeMinuteAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        NodeAnalysisVerfiy.INSTANCE.verfiyPersistenceService(recordDataAnswer.recordObj.getRecordDataMap(), 201703310915L);
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/noderef/analysis/NodeRefDayAnalysisTestCase.java

+package com.a.eye.skywalking.collector.worker.noderef.analysis;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.WorkerRefs;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.RecordDataAnswer;
+import com.a.eye.skywalking.collector.worker.noderef.persistence.NodeRefDayAgg;
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.segment.mock.SegmentMock;
+import com.a.eye.skywalking.collector.worker.storage.RecordData;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClusterWorkerContext.class})
+@PowerMockIgnore({"javax.management.*"})
+public class NodeRefDayAnalysisTestCase {
+
+    private NodeRefDayAnalysis nodeRefDayAnalysis;
+    private SegmentMock segmentMock = new SegmentMock();
+    private RecordDataAnswer recordDataAnswer;
+
+    @Before
+    public void init() throws Exception {
+        ClusterWorkerContext clusterWorkerContext = PowerMockito.mock(ClusterWorkerContext.class);
+        WorkerRefs workerRefs = mock(WorkerRefs.class);
+        recordDataAnswer = new RecordDataAnswer();
+        doAnswer(recordDataAnswer).when(workerRefs).tell(Mockito.any(RecordData.class));
+
+        when(clusterWorkerContext.lookup(NodeRefDayAgg.Role.INSTANCE)).thenReturn(workerRefs);
+
+        LocalWorkerContext localWorkerContext = PowerMockito.mock(LocalWorkerContext.class);
+        WorkerRefs nodeRefResSumWorkerRefs = mock(WorkerRefs.class);
+        when(localWorkerContext.lookup(NodeRefResSumDayAnalysis.Role.INSTANCE)).thenReturn(nodeRefResSumWorkerRefs);
+        nodeRefDayAnalysis = new NodeRefDayAnalysis(NodeRefDayAnalysis.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(NodeRefDayAnalysis.class.getSimpleName(), NodeRefDayAnalysis.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), NodeRefDayAnalysis.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(NodeRefDayAnalysis.class.getSimpleName(), NodeRefDayAnalysis.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(NodeRefDayAnalysis.class.getSimpleName(), NodeRefDayAnalysis.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Node.NodeRefDayAnalysis.Size = testSize;
+        Assert.assertEquals(testSize, NodeRefDayAnalysis.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testCacheServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockCacheServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeRefDayAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        Map<String, RecordData> recordDataMap = recordDataAnswer.recordObj.getRecordDataMap();
+        for (Map.Entry<String, RecordData> entry : recordDataMap.entrySet()) {
+            String id = entry.getKey();
+            RecordData recordData = entry.getValue();
+            System.out.println(id);
+            System.out.println(recordData.getRecord().toString());
+        }
+    }
+
+    @Test
+    public void testPortalServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPortalServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeRefDayAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        Map<String, RecordData> recordDataMap = recordDataAnswer.recordObj.getRecordDataMap();
+        for (Map.Entry<String, RecordData> entry : recordDataMap.entrySet()) {
+            String id = entry.getKey();
+            RecordData recordData = entry.getValue();
+            System.out.println(id);
+            System.out.println(recordData.getRecord().toString());
+        }
+    }
+
+    @Test
+    public void testPersistenceServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPersistenceServiceSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeRefDayAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        Map<String, RecordData> recordDataMap = recordDataAnswer.recordObj.getRecordDataMap();
+        for (Map.Entry<String, RecordData> entry : recordDataMap.entrySet()) {
+            String id = entry.getKey();
+            RecordData recordData = entry.getValue();
+            System.out.println(id);
+            System.out.println(recordData.getRecord().toString());
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/noderef/analysis/NodeRefMinuteAnalysisTestCase.java

+package com.a.eye.skywalking.collector.worker.noderef.analysis;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.WorkerRefs;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.RecordDataAnswer;
+import com.a.eye.skywalking.collector.worker.noderef.persistence.NodeRefMinuteAgg;
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.segment.mock.SegmentMock;
+import com.a.eye.skywalking.collector.worker.storage.RecordData;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClusterWorkerContext.class})
+@PowerMockIgnore({"javax.management.*"})
+public class NodeRefMinuteAnalysisTestCase {
+
+    private NodeRefMinuteAnalysis nodeRefMinuteAnalysis;
+    private SegmentMock segmentMock = new SegmentMock();
+    private RecordDataAnswer recordDataAnswer;
+
+    @Before
+    public void init() throws Exception {
+        ClusterWorkerContext clusterWorkerContext = PowerMockito.mock(ClusterWorkerContext.class);
+        WorkerRefs workerRefs = mock(WorkerRefs.class);
+        recordDataAnswer = new RecordDataAnswer();
+        doAnswer(recordDataAnswer).when(workerRefs).tell(Mockito.any(RecordData.class));
+
+        when(clusterWorkerContext.lookup(NodeRefMinuteAgg.Role.INSTANCE)).thenReturn(workerRefs);
+
+        LocalWorkerContext localWorkerContext = PowerMockito.mock(LocalWorkerContext.class);
+        WorkerRefs nodeRefResSumWorkerRefs = mock(WorkerRefs.class);
+        when(localWorkerContext.lookup(NodeRefResSumMinuteAnalysis.Role.INSTANCE)).thenReturn(nodeRefResSumWorkerRefs);
+        nodeRefMinuteAnalysis = new NodeRefMinuteAnalysis(NodeRefMinuteAnalysis.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(NodeRefMinuteAnalysis.class.getSimpleName(), NodeRefMinuteAnalysis.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), NodeRefMinuteAnalysis.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(NodeRefMinuteAnalysis.class.getSimpleName(), NodeRefMinuteAnalysis.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(NodeRefMinuteAnalysis.class.getSimpleName(), NodeRefMinuteAnalysis.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Node.NodeRefMinuteAnalysis.Size = testSize;
+        Assert.assertEquals(testSize, NodeRefMinuteAnalysis.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testCacheServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockCacheServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeRefMinuteAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        Map<String, RecordData> recordDataMap = recordDataAnswer.recordObj.getRecordDataMap();
+        for (Map.Entry<String, RecordData> entry : recordDataMap.entrySet()) {
+            String id = entry.getKey();
+            RecordData recordData = entry.getValue();
+            System.out.println(id);
+            System.out.println(recordData.getRecord().toString());
+        }
+    }
+
+    @Test
+    public void testPortalServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPortalServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeRefMinuteAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        Map<String, RecordData> recordDataMap = recordDataAnswer.recordObj.getRecordDataMap();
+        for (Map.Entry<String, RecordData> entry : recordDataMap.entrySet()) {
+            String id = entry.getKey();
+            RecordData recordData = entry.getValue();
+            System.out.println(id);
+            System.out.println(recordData.getRecord().toString());
+        }
+    }
+
+    @Test
+    public void testPersistenceServiceAnalyse() throws Exception {
+        WorkerConfig.Analysis.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPersistenceServiceSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            nodeRefMinuteAnalysis.analyse(segmentWithTimeSlice);
+        }
+
+        Map<String, RecordData> recordDataMap = recordDataAnswer.recordObj.getRecordDataMap();
+        for (Map.Entry<String, RecordData> entry : recordDataMap.entrySet()) {
+            String id = entry.getKey();
+            RecordData recordData = entry.getValue();
+            System.out.println(id);
+            System.out.println(recordData.getRecord().toString());
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/segment/SegmentPostTestCase.java

@@ -16,6 +16,8 @@ import com.a.eye.skywalking.collector.worker.segment.persistence.SegmentExceptio
 import com.a.eye.skywalking.collector.worker.segment.persistence.SegmentSave;
 import com.a.eye.skywalking.collector.worker.tools.DateTools;
 import com.google.gson.JsonObject;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -38,6 +40,8 @@ import static org.powermock.api.mockito.PowerMockito.*;
 @PowerMockIgnore({"javax.management.*"})
 public class SegmentPostTestCase {
 
+    private Logger logger = LogManager.getFormatterLogger(SegmentPostTestCase.class);
+
     private SegmentMock segmentMock;
     private SegmentPost segmentPost;
     private LocalWorkerContext localWorkerContext;
@@ -231,6 +235,7 @@ public class SegmentPostTestCase {
         @Override
         public Object answer(InvocationOnMock invocation) throws Throwable {
             JsonObject jsonObject = (JsonObject) invocation.getArguments()[0];
+            logger.info("SegmentSave json: " + jsonObject.toString());
             minute = jsonObject.get("minute").getAsLong();
             hour = jsonObject.get("hour").getAsLong();
             day = jsonObject.get("day").getAsLong();

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/segment/SegmentSaveTestCase.java

-package com.a.eye.skywalking.collector.worker.segment;
-
-import com.a.eye.skywalking.collector.worker.segment.persistence.SegmentSave;
-import com.google.gson.JsonObject;
-import org.junit.Test;
-import org.mockito.Mockito;
-
-/**
- * @author pengys5
- */
-public class SegmentSaveTestCase {
-
-    @Test
-    public void testAnalyse() throws Exception{
-        SegmentSave segmentSave = Mockito.mock(SegmentSave.class);
-
-        JsonObject segment_1 = new JsonObject();
-        segmentSave.analyse(segment_1);
-    }
-}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/segment/mock/SegmentMock.java

 package com.a.eye.skywalking.collector.worker.segment.mock;
 
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.tools.DateTools;
 import com.a.eye.skywalking.collector.worker.tools.JsonFileReader;
+import com.a.eye.skywalking.trace.SegmentsMessage;
+import com.a.eye.skywalking.trace.TraceSegment;
+import com.google.gson.Gson;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.io.FileNotFoundException;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * @author pengys5
  */
 public class SegmentMock {
 
+    private Logger logger = LogManager.getFormatterLogger(SegmentMock.class);
+
+    private Gson gson = new Gson();
     private String path = this.getClass().getResource("/").getPath();
 
     private final String CacheServiceJsonFile = path + "/json/segment/post/normal/cache-service.json";
     private final String PersistenceServiceJsonFile = path + "/json/segment/post/normal/persistence-service.json";
     private final String PortalServiceJsonFile = path + "/json/segment/post/normal/portal-service.json";
 
+    private final String CacheServiceExceptionJsonFile = path + "/json/segment/post/exception/cache-service.json";
+    private final String PortalServiceExceptionJsonFile = path + "/json/segment/post/exception/portal-service.json";
+
+    private final String SpecialJsonFile = path + "/json/segment/post/special/special.json";
+
+    public String mockSpecialSegmentAsString() throws FileNotFoundException {
+        return JsonFileReader.INSTANCE.read(SpecialJsonFile);
+    }
+
     public String mockCacheServiceSegmentAsString() throws FileNotFoundException {
         return JsonFileReader.INSTANCE.read(CacheServiceJsonFile);
     }
@@ -26,4 +47,70 @@ public class SegmentMock {
     public String mockPortalServiceSegmentAsString() throws FileNotFoundException {
         return JsonFileReader.INSTANCE.read(PortalServiceJsonFile);
     }
+
+    public String mockCacheServiceExceptionSegmentAsString() throws FileNotFoundException {
+        return JsonFileReader.INSTANCE.read(CacheServiceExceptionJsonFile);
+    }
+
+    public String mockPortalServiceExceptionSegmentAsString() throws FileNotFoundException {
+        return JsonFileReader.INSTANCE.read(PortalServiceExceptionJsonFile);
+    }
+
+    public List<SegmentPost.SegmentWithTimeSlice> mockSpecialSegmentTimeSlice() throws FileNotFoundException {
+        String specialSegmentAsString = mockSpecialSegmentAsString();
+        logger.debug(specialSegmentAsString);
+        return createSegmentWithTimeSliceList(specialSegmentAsString);
+    }
+
+    public List<SegmentPost.SegmentWithTimeSlice> mockCacheServiceExceptionSegmentTimeSlice() throws FileNotFoundException {
+        String cacheServiceExceptionSegmentAsString = mockCacheServiceExceptionSegmentAsString();
+        logger.debug(cacheServiceExceptionSegmentAsString);
+        return createSegmentWithTimeSliceList(cacheServiceExceptionSegmentAsString);
+    }
+
+    public List<SegmentPost.SegmentWithTimeSlice> mockPortalServiceExceptionSegmentTimeSlice() throws FileNotFoundException {
+        String portalServiceExceptionSegmentAsString = mockPortalServiceExceptionSegmentAsString();
+        logger.debug(portalServiceExceptionSegmentAsString);
+        return createSegmentWithTimeSliceList(portalServiceExceptionSegmentAsString);
+    }
+
+    public List<SegmentPost.SegmentWithTimeSlice> mockCacheServiceSegmentSegmentTimeSlice() throws FileNotFoundException {
+        String cacheServiceSegmentAsString = mockCacheServiceSegmentAsString();
+        logger.debug(cacheServiceSegmentAsString);
+        return createSegmentWithTimeSliceList(cacheServiceSegmentAsString);
+    }
+
+    public List<SegmentPost.SegmentWithTimeSlice> mockPersistenceServiceSegmentTimeSlice() throws FileNotFoundException {
+        String persistenceServiceSegmentAsString = mockPersistenceServiceSegmentAsString();
+        logger.debug(persistenceServiceSegmentAsString);
+        return createSegmentWithTimeSliceList(persistenceServiceSegmentAsString);
+    }
+
+    public List<SegmentPost.SegmentWithTimeSlice> mockPortalServiceSegmentSegmentTimeSlice() throws FileNotFoundException {
+        String portalServiceSegmentAsString = mockPortalServiceSegmentAsString();
+        logger.debug(portalServiceSegmentAsString);
+        return createSegmentWithTimeSliceList(portalServiceSegmentAsString);
+    }
+
+    private List<SegmentPost.SegmentWithTimeSlice> createSegmentWithTimeSliceList(String segmentJsonStr) {
+        SegmentsMessage segmentsMessage = gson.fromJson(segmentJsonStr, SegmentsMessage.class);
+        List<TraceSegment> segmentList = segmentsMessage.getSegments();
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = new ArrayList<>();
+        for (TraceSegment newSegment : segmentList) {
+            SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice = createSegmentWithTimeSlice(newSegment);
+            segmentWithTimeSliceList.add(segmentWithTimeSlice);
+        }
+        return segmentWithTimeSliceList;
+    }
+
+    private SegmentPost.SegmentWithTimeSlice createSegmentWithTimeSlice(TraceSegment newSegment) {
+        long minuteSlice = DateTools.getMinuteSlice(newSegment.getStartTime());
+        long hourSlice = DateTools.getHourSlice(newSegment.getStartTime());
+        long daySlice = DateTools.getDaySlice(newSegment.getStartTime());
+        int second = DateTools.getSecond(newSegment.getStartTime());
+
+        SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice = new SegmentPost.SegmentWithTimeSlice(newSegment, minuteSlice, hourSlice, daySlice, second);
+        return segmentWithTimeSlice;
+    }
 }

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/segment/persistence/SegmentCostSaveTestCase.java

+package com.a.eye.skywalking.collector.worker.segment.persistence;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.MockEsBulkClient;
+import com.a.eye.skywalking.collector.worker.mock.SaveToEsSourceAnswer;
+import com.a.eye.skywalking.collector.worker.segment.SegmentCostIndex;
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.segment.mock.SegmentMock;
+import com.a.eye.skywalking.collector.worker.storage.EsClient;
+import com.a.eye.skywalking.collector.worker.tools.DateTools;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import org.apache.logging.log4j.LogManager;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.List;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({EsClient.class, LogManager.class})
+@PowerMockIgnore({"javax.management.*"})
+public class SegmentCostSaveTestCase {
+
+    private SegmentCostSave segmentCostSave;
+    private SegmentMock segmentMock = new SegmentMock();
+    private MockEsBulkClient mockEsBulkClient = new MockEsBulkClient();
+
+    private SaveToEsSourceAnswer saveToEsSourceAnswer;
+
+    @Before
+    public void init() throws Exception {
+        mockEsBulkClient.createMock();
+
+        saveToEsSourceAnswer = new SaveToEsSourceAnswer();
+        when(mockEsBulkClient.indexRequestBuilder.setSource(Mockito.anyString())).thenAnswer(saveToEsSourceAnswer);
+
+        ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
+        LocalWorkerContext localWorkerContext = new LocalWorkerContext();
+        segmentCostSave = new SegmentCostSave(SegmentCostSave.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testEsIndex() {
+        Assert.assertEquals(SegmentCostIndex.Index, segmentCostSave.esIndex());
+    }
+
+    @Test
+    public void testEsType() {
+        Assert.assertEquals(SegmentCostIndex.Type_Record, segmentCostSave.esType());
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(SegmentCostSave.class.getSimpleName(), SegmentCostSave.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), SegmentCostSave.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(SegmentCostSave.class.getSimpleName(), SegmentCostSave.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(SegmentCostSave.class.getSimpleName(), SegmentCostSave.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Segment.SegmentCostSave.Size = testSize;
+        Assert.assertEquals(testSize, SegmentCostSave.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testPersistenceServiceAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPersistenceServiceSegmentTimeSlice();
+
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            segmentCostSave.analyse(segmentWithTimeSlice);
+        }
+
+        JsonArray sourceArray = saveToEsSourceAnswer.sourceObj.getSource();
+        Assert.assertEquals(1, sourceArray.size());
+
+        JsonObject costJsonObj = sourceArray.get(0).getAsJsonObject();
+        Assert.assertEquals("Segment.1490922929274.1382198130.5997.47.1", costJsonObj.get("segId").getAsString());
+        Assert.assertEquals(1490922929274L, costJsonObj.get("startTime").getAsLong());
+        Assert.assertEquals(1490922929288L, costJsonObj.get("EndTime").getAsLong());
+        Assert.assertEquals("/persistence/query", costJsonObj.get("operationName").getAsString());
+        Assert.assertEquals(DateTools.changeToUTCSlice(201703310915L), costJsonObj.get("timeSlice").getAsLong());
+        Assert.assertEquals(14, costJsonObj.get("cost").getAsInt());
+    }
+
+    @Test
+    public void testCacheServiceAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 2;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockCacheServiceSegmentSegmentTimeSlice();
+
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            segmentCostSave.analyse(segmentWithTimeSlice);
+        }
+
+        JsonArray sourceArray = saveToEsSourceAnswer.sourceObj.getSource();
+        Assert.assertEquals(2, sourceArray.size());
+
+        JsonObject costJsonObj_0 = null;
+        JsonObject costJsonObj_1 = null;
+        for (int i = 0; i < sourceArray.size(); i++) {
+            JsonObject costJsonObj = sourceArray.get(i).getAsJsonObject();
+            if (costJsonObj.get("segId").getAsString().equals("Segment.1490922929258.927784221.5991.27.1")) {
+                costJsonObj_0 = costJsonObj;
+            } else if (costJsonObj.get("segId").getAsString().equals("Segment.1490922929298.927784221.5991.28.1")) {
+                costJsonObj_1 = costJsonObj;
+            }
+        }
+
+        Assert.assertEquals(1490922929258L, costJsonObj_0.get("startTime").getAsLong());
+        Assert.assertEquals(1490922929261L, costJsonObj_0.get("EndTime").getAsLong());
+        Assert.assertEquals("com.a.eye.skywalking.test.cache.CacheService.findCache(java.lang.String)", costJsonObj_0.get("operationName").getAsString());
+        Assert.assertEquals(DateTools.changeToUTCSlice(201703310915L), costJsonObj_0.get("timeSlice").getAsLong());
+        Assert.assertEquals(3, costJsonObj_0.get("cost").getAsInt());
+
+        Assert.assertEquals(1490922929298L, costJsonObj_1.get("startTime").getAsLong());
+        Assert.assertEquals(1490922929303L, costJsonObj_1.get("EndTime").getAsLong());
+        Assert.assertEquals("com.a.eye.skywalking.test.cache.CacheService.updateCache(java.lang.String,java.lang.String)", costJsonObj_1.get("operationName").getAsString());
+        Assert.assertEquals(DateTools.changeToUTCSlice(201703310915L), costJsonObj_1.get("timeSlice").getAsLong());
+        Assert.assertEquals(5, costJsonObj_1.get("cost").getAsInt());
+    }
+
+    @Test
+    public void testPortalServiceAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPortalServiceSegmentSegmentTimeSlice();
+
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            segmentCostSave.analyse(segmentWithTimeSlice);
+        }
+
+        JsonArray sourceArray = saveToEsSourceAnswer.sourceObj.getSource();
+        Assert.assertEquals(1, sourceArray.size());
+
+        JsonObject costJsonObj = sourceArray.get(0).getAsJsonObject();
+        Assert.assertEquals("Segment.1490922929254.1797892356.6003.69.1", costJsonObj.get("segId").getAsString());
+        Assert.assertEquals(1490922929254L, costJsonObj.get("startTime").getAsLong());
+        Assert.assertEquals(1490922929306L, costJsonObj.get("EndTime").getAsLong());
+        Assert.assertEquals("/portal/", costJsonObj.get("operationName").getAsString());
+        Assert.assertEquals(DateTools.changeToUTCSlice(201703310915L), costJsonObj.get("timeSlice").getAsLong());
+        Assert.assertEquals(52, costJsonObj.get("cost").getAsInt());
+    }
+
+    @Test
+    public void testCacheServiceExceptionAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockCacheServiceExceptionSegmentTimeSlice();
+
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            segmentCostSave.analyse(segmentWithTimeSlice);
+        }
+
+        JsonArray sourceArray = saveToEsSourceAnswer.sourceObj.getSource();
+        Assert.assertEquals(1, sourceArray.size());
+
+        JsonObject costJsonObj = sourceArray.get(0).getAsJsonObject();
+        Assert.assertEquals("Segment.1490923010328.927784221.5991.32.1", costJsonObj.get("segId").getAsString());
+        Assert.assertEquals(1490923010328L, costJsonObj.get("startTime").getAsLong());
+        Assert.assertEquals(1490923010329L, costJsonObj.get("EndTime").getAsLong());
+        Assert.assertEquals("com.a.eye.skywalking.test.cache.CacheService.findCacheWithException(java.lang.String)", costJsonObj.get("operationName").getAsString());
+        Assert.assertEquals(DateTools.changeToUTCSlice(201703310916L), costJsonObj.get("timeSlice").getAsLong());
+        Assert.assertEquals(1, costJsonObj.get("cost").getAsInt());
+    }
+
+    @Test
+    public void testPortalServiceExceptionAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> segmentWithTimeSliceList = segmentMock.mockPortalServiceExceptionSegmentTimeSlice();
+
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : segmentWithTimeSliceList) {
+            segmentCostSave.analyse(segmentWithTimeSlice);
+        }
+
+        JsonArray sourceArray = saveToEsSourceAnswer.sourceObj.getSource();
+        Assert.assertEquals(1, sourceArray.size());
+
+        JsonObject costJsonObj = sourceArray.get(0).getAsJsonObject();
+        Assert.assertEquals("Segment.1490923010324.1797892356.6003.67.1", costJsonObj.get("segId").getAsString());
+        Assert.assertEquals(1490923010324L, costJsonObj.get("startTime").getAsLong());
+        Assert.assertEquals(1490923010336L, costJsonObj.get("EndTime").getAsLong());
+        Assert.assertEquals("/portal/cache/exception/test", costJsonObj.get("operationName").getAsString());
+        Assert.assertEquals(DateTools.changeToUTCSlice(201703310916L), costJsonObj.get("timeSlice").getAsLong());
+        Assert.assertEquals(12, costJsonObj.get("cost").getAsInt());
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/segment/persistence/SegmentExceptionSaveTestCase.java

+package com.a.eye.skywalking.collector.worker.segment.persistence;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.MockEsBulkClient;
+import com.a.eye.skywalking.collector.worker.segment.SegmentExceptionIndex;
+import com.a.eye.skywalking.collector.worker.segment.SegmentPost;
+import com.a.eye.skywalking.collector.worker.segment.mock.SegmentMock;
+import com.a.eye.skywalking.collector.worker.storage.EsClient;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({EsClient.class})
+@PowerMockIgnore({"javax.management.*"})
+public class SegmentExceptionSaveTestCase {
+
+    private Logger logger = LogManager.getFormatterLogger(SegmentExceptionSaveTestCase.class);
+
+    private SegmentMock segmentMock = new SegmentMock();
+    private SegmentExceptionSave segmentExceptionSave;
+    private SaveToEsSource saveToEsSource;
+    private MockEsBulkClient mockEsBulkClient = new MockEsBulkClient();
+
+    @Before
+    public void init() {
+        mockEsBulkClient.createMock();
+
+        saveToEsSource = new SaveToEsSource();
+        when(mockEsBulkClient.indexRequestBuilder.setSource(Mockito.anyString())).thenAnswer(saveToEsSource);
+
+        ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
+        LocalWorkerContext localWorkerContext = new LocalWorkerContext();
+
+        segmentExceptionSave = new SegmentExceptionSave(SegmentExceptionSave.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testEsIndex() {
+        Assert.assertEquals(SegmentExceptionIndex.Index, segmentExceptionSave.esIndex());
+    }
+
+    @Test
+    public void testEsType() {
+        Assert.assertEquals(SegmentExceptionIndex.Type_Record, segmentExceptionSave.esType());
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(SegmentExceptionSave.class.getSimpleName(), SegmentExceptionSave.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), SegmentExceptionSave.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(SegmentExceptionSave.class.getSimpleName(), SegmentExceptionSave.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(SegmentExceptionSave.class.getSimpleName(), SegmentExceptionSave.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Segment.SegmentExceptionSave.Size = testSize;
+        Assert.assertEquals(testSize, SegmentExceptionSave.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testSuccessAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> cacheServiceList = segmentMock.mockCacheServiceSegmentSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : cacheServiceList) {
+            segmentExceptionSave.analyse(segmentWithTimeSlice);
+        }
+
+        JsonObject isError_1 = saveToEsSource.isErrorMap.get("Segment.1490922929258.927784221.5991.27.1");
+        Assert.assertEquals(false, isError_1.get("isError").getAsBoolean());
+        Assert.assertEquals(0, isError_1.get("errorKind").getAsJsonArray().size());
+
+        JsonObject isError_2 = saveToEsSource.isErrorMap.get("Segment.1490922929298.927784221.5991.28.1");
+        Assert.assertEquals(false, isError_2.get("isError").getAsBoolean());
+        Assert.assertEquals(0, isError_2.get("errorKind").getAsJsonArray().size());
+    }
+
+    @Test
+    public void testErrorAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 1;
+
+        List<SegmentPost.SegmentWithTimeSlice> cacheServiceList = segmentMock.mockCacheServiceExceptionSegmentTimeSlice();
+        for (SegmentPost.SegmentWithTimeSlice segmentWithTimeSlice : cacheServiceList) {
+            segmentExceptionSave.analyse(segmentWithTimeSlice);
+        }
+
+        JsonObject isError_1 = saveToEsSource.isErrorMap.get("Segment.1490923010328.927784221.5991.32.1");
+        Assert.assertEquals(true, isError_1.get("isError").getAsBoolean());
+        Assert.assertEquals("com.weibo.api.motan.exception.MotanBizException", isError_1.get("errorKind").getAsJsonArray().get(0).getAsString());
+    }
+
+    class SaveToEsSource implements Answer<Object> {
+
+        Map<String, JsonObject> isErrorMap = new HashMap<>();
+
+        @Override
+        public Object answer(InvocationOnMock invocation) throws Throwable {
+            Gson gson = new Gson();
+            String source = (String) invocation.getArguments()[0];
+            JsonObject sourceJsonObj = gson.fromJson(source, JsonObject.class);
+            logger.info("es source: %s", sourceJsonObj.toString());
+            isErrorMap.put(sourceJsonObj.get("segId").getAsString(), sourceJsonObj);
+            return null;
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/segment/persistence/SegmentSaveTestCase.java

+package com.a.eye.skywalking.collector.worker.segment.persistence;
+
+import com.a.eye.skywalking.collector.actor.ClusterWorkerContext;
+import com.a.eye.skywalking.collector.actor.LocalWorkerContext;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.mock.MockEsBulkClient;
+import com.a.eye.skywalking.collector.worker.segment.SegmentCostIndex;
+import com.a.eye.skywalking.collector.worker.segment.SegmentIndex;
+import com.a.eye.skywalking.collector.worker.storage.EsClient;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({EsClient.class})
+@PowerMockIgnore({"javax.management.*"})
+public class SegmentSaveTestCase {
+
+    private SegmentSave segmentSave;
+    private SaveToEsSource saveToEsSource;
+    private MockEsBulkClient mockEsBulkClient = new MockEsBulkClient();
+
+    @Before
+    public void init() {
+        mockEsBulkClient.createMock();
+
+        saveToEsSource = new SaveToEsSource();
+        when(mockEsBulkClient.indexRequestBuilder.setSource(Mockito.anyString())).thenAnswer(saveToEsSource);
+
+        ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
+        LocalWorkerContext localWorkerContext = new LocalWorkerContext();
+
+        segmentSave = new SegmentSave(SegmentSave.Role.INSTANCE, clusterWorkerContext, localWorkerContext);
+    }
+
+    @Test
+    public void testEsIndex() {
+        Assert.assertEquals(SegmentIndex.Index, segmentSave.esIndex());
+    }
+
+    @Test
+    public void testEsType() {
+        Assert.assertEquals(SegmentIndex.Type_Record, segmentSave.esType());
+    }
+
+    @Test
+    public void testRole() {
+        Assert.assertEquals(SegmentSave.class.getSimpleName(), SegmentSave.Role.INSTANCE.roleName());
+        Assert.assertEquals(RollingSelector.class.getSimpleName(), SegmentSave.Role.INSTANCE.workerSelector().getClass().getSimpleName());
+    }
+
+    @Test
+    public void testFactory() {
+        Assert.assertEquals(SegmentSave.class.getSimpleName(), SegmentSave.Factory.INSTANCE.role().roleName());
+        Assert.assertEquals(SegmentSave.class.getSimpleName(), SegmentSave.Factory.INSTANCE.workerInstance(null).getClass().getSimpleName());
+
+        int testSize = 10;
+        WorkerConfig.Queue.Segment.SegmentSave.Size = testSize;
+        Assert.assertEquals(testSize, SegmentSave.Factory.INSTANCE.queueSize());
+    }
+
+    @Test
+    public void testAnalyse() throws Exception {
+        WorkerConfig.Persistence.Data.size = 1;
+
+        JsonObject segment_1 = new JsonObject();
+        segment_1.addProperty("ts", "segment_1");
+        segmentSave.analyse(segment_1);
+
+        Assert.assertEquals("segment_1", saveToEsSource.ts);
+    }
+
+    class SaveToEsSource implements Answer<Object> {
+
+        String ts = "";
+
+        @Override
+        public Object answer(InvocationOnMock invocation) throws Throwable {
+            Gson gson = new Gson();
+            String source = (String) invocation.getArguments()[0];
+            JsonObject sourceJsonObj = gson.fromJson(source, JsonObject.class);
+            ts = sourceJsonObj.get("ts").getAsString();
+            return null;
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/resources/json/segment/post/special/special.json

+[
+  {
+    "ts": "Segment.1491277162066.18986177.70531.27.1",
+    "st": 1491277162066,
+    "et": 1491277165743,
+    "rs": [
+      {
+        "ts": "Segment.1491277147441.-1562443425.70539.65.1",
+        "si": 6,
+        "ac": "portal-service",
+        "ph": "127.0.0.1:8002"
+      }
+    ],
+    "ss": [
+      {
+        "si": 2,
+        "ps": 1,
+        "st": 1491277162141,
+        "et": 1491277162144,
+        "on": "Jedis/getClient",
+        "ts": {
+          "span.layer": "db",
+          "component": "Redis",
+          "db.type": "Redis",
+          "peer.host": "127.0.0.1",
+          "span.kind": "client"
+        },
+        "tb": {},
+        "ti": {
+          "peer.port": 6379
+        },
+        "lo": []
+      },
+      {
+        "si": 3,
+        "ps": 1,
+        "st": 1491277162144,
+        "et": 1491277162147,
+        "on": "Jedis/getClient",
+        "ts": {
+          "span.layer": "db",
+          "component": "Redis",
+          "db.type": "Redis",
+          "peer.host": "127.0.0.1",
+          "span.kind": "client"
+        },
+        "tb": {},
+        "ti": {
+          "peer.port": 6379
+        },
+        "lo": []
+      },
+      {
+        "si": 4,
+        "ps": 1,
+        "st": 1491277162147,
+        "et": 1491277162147,
+        "on": "Jedis/isConnected",
+        "ts": {
+          "span.layer": "db",
+          "component": "Redis",
+          "db.type": "Redis",
+          "peer.host": "127.0.0.1",
+          "span.kind": "client"
+        },
+        "tb": {},
+        "ti": {
+          "peer.port": 6379
+        },
+        "lo": []
+      },
+      {
+        "si": 5,
+        "ps": 1,
+        "st": 1491277162147,
+        "et": 1491277164410,
+        "on": "Jedis/ping",
+        "ts": {
+          "span.layer": "db",
+          "component": "Redis",
+          "db.type": "Redis",
+          "peer.host": "127.0.0.1",
+          "span.kind": "client"
+        },
+        "tb": {},
+        "ti": {
+          "peer.port": 6379
+        },
+        "lo": []
+      },
+      {
+        "si": 6,
+        "ps": 1,
+        "st": 1491277164420,
+        "et": 1491277165738,
+        "on": "Jedis/set",
+        "ts": {
+          "span.layer": "db",
+          "component": "Redis",
+          "db.type": "Redis",
+          "peer.host": "127.0.0.1",
+          "span.kind": "client",
+          "db.statement": "set test_NEW"
+        },
+        "tb": {},
+        "ti": {
+          "peer.port": 6379
+        },
+        "lo": []
+      },
+      {
+        "si": 7,
+        "ps": 1,
+        "st": 1491277165741,
+        "et": 1491277165743,
+        "on": "H2/JDBI/PreparedStatement/executeUpdate",
+        "ts": {
+          "db.instance": "dataSource",
+          "span.layer": "db",
+          "db.type": "sql",
+          "component": "H2",
+          "span.kind": "client",
+          "db.statement": "INSERT INTO CACHE_TABLE(CACHE_VALUE, CACHE_KEY) VALUES(?, ?)",
+          "peer.host": "localhost"
+        },
+        "tb": {},
+        "ti": {
+          "peer.port": -1
+        },
+        "lo": []
+      },
+      {
+        "si": 8,
+        "ps": 1,
+        "st": 1491277165743,
+        "et": 1491277165743,
+        "on": "H2/JDBI/Connection/close",
+        "ts": {
+          "db.instance": "dataSource",
+          "span.layer": "db",
+          "db.type": "sql",
+          "component": "H2",
+          "span.kind": "client",
+          "db.statement": "",
+          "peer.host": "localhost"
+        },
+        "tb": {},
+        "ti": {
+          "peer.port": -1
+        },
+        "lo": []
+      },
+      {
+        "si": 1,
+        "ps": 0,
+        "st": 1491277162066,
+        "et": 1491277165743,
+        "on": "Motan_default_rpc_com.a.eye.skywalking.test.cache.CacheService.updateCache(java.lang.String,java.lang.String)",
+        "ts": {
+          "requestId": "1563717428890828802"
+        },
+        "tb": {},
+        "ti": {},
+        "lo": []
+      },
+      {
+        "si": 0,
+        "ps": -1,
+        "st": 1491277162066,
+        "et": 1491277165743,
+        "on": "com.a.eye.skywalking.test.cache.CacheService.updateCache(java.lang.String,java.lang.String)",
+        "ts": {
+          "span.layer": "rpc",
+          "component": "Motan",
+          "span.kind": "server"
+        },
+        "tb": {},
+        "ti": {},
+        "lo": []
+      }
+    ],
+    "ac": "cache-service",
+    "gt": [
+      "Trace.1491277147443.-1562443425.70539.65.2"
+    ]
+  }
+]
\ No newline at end of file