worker storage mock finish

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

@@ -22,8 +22,8 @@ public class CollectorBootStartUp {
 
         CollectorSystem collectorSystem = new CollectorSystem();
         collectorSystem.boot();
-        EsClient.boot();
-//        IndexCreator.INSTANCE.create();
+        EsClient.INSTANCE.boot();
+        IndexCreator.INSTANCE.create();
         HttpServer.INSTANCE.boot((ClusterWorkerContext) collectorSystem.getClusterContext());
     }
 }

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

@@ -61,7 +61,7 @@ public abstract class MergePersistenceMember extends PersistenceMember {
     }
 
     private MultiGetResponse searchFromEs() {
-        Client client = EsClient.getClient();
+        Client client = EsClient.INSTANCE.getClient();
         MultiGetRequestBuilder multiGetRequestBuilder = client.prepareMultiGet();
 
         Iterator<Map.Entry<String, MergeData>> iterator = persistenceData.iterator();
@@ -74,7 +74,7 @@ public abstract class MergePersistenceMember extends PersistenceMember {
     }
 
     private boolean saveToEs() {
-        Client client = EsClient.getClient();
+        Client client = EsClient.INSTANCE.getClient();
         BulkRequestBuilder bulkRequest = client.prepareBulk();
         logger.debug("persistenceData size: %s", persistenceData.size());
 

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

@@ -61,7 +61,7 @@ public abstract class MetricPersistenceMember extends PersistenceMember {
     }
 
     private MultiGetResponse searchFromEs() {
-        Client client = EsClient.getClient();
+        Client client = EsClient.INSTANCE.getClient();
         MultiGetRequestBuilder multiGetRequestBuilder = client.prepareMultiGet();
 
         Iterator<Map.Entry<String, MetricData>> iterator = persistenceData.iterator();
@@ -75,7 +75,7 @@ public abstract class MetricPersistenceMember extends PersistenceMember {
     }
 
     private boolean saveToEs() {
-        Client client = EsClient.getClient();
+        Client client = EsClient.INSTANCE.getClient();
         BulkRequestBuilder bulkRequest = client.prepareBulk();
         logger.debug("persistenceData size: %s", persistenceData.size());
 

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

@@ -55,7 +55,7 @@ public abstract class RecordPersistenceMember extends PersistenceMember {
     }
 
     private boolean saveToEs() {
-        Client client = EsClient.getClient();
+        Client client = EsClient.INSTANCE.getClient();
         BulkRequestBuilder bulkRequest = client.prepareBulk();
         logger.debug("persistenceData size: %s", getPersistenceData().size());
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/globaltrace/persistence/GlobalTraceSearchWithGlobalId.java

@@ -40,7 +40,7 @@ public class GlobalTraceSearchWithGlobalId extends AbstractLocalSyncWorker {
     @Override
     protected void onWork(Object request, Object response) throws Exception {
         if (request instanceof String) {
-            Client client = EsClient.getClient();
+            Client client = EsClient.INSTANCE.getClient();
             String globalId = (String) request;
             String globalTraceData = client.prepareGet(GlobalTraceIndex.Index, GlobalTraceIndex.Type_Record, globalId).get().getSourceAsString();
             JsonObject globalTraceObj = gson.fromJson(globalTraceData, JsonObject.class);

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

@@ -36,7 +36,7 @@ abstract class AbstractNodeCompAnalysis extends RecordAnalysisMember {
             for (Span span : spanList) {
                 String kind = Tags.SPAN_KIND.get(span);
                 if (Tags.SPAN_KIND_CLIENT.equals(kind) && ClientSpanIsLeafTools.isLeaf(span.getSpanId(), spanList)) {
-                    String peers = SpanPeersTools.getPeers(span);
+                    String peers = SpanPeersTools.INSTANCE.getPeers(span);
 
                     JsonObject compJsonObj = new JsonObject();
                     compJsonObj.addProperty(NodeCompIndex.Peers, peers);

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

@@ -27,7 +27,7 @@ public class NodeCompLoad extends AbstractLocalSyncWorker {
 
     @Override
     public void onWork(Object request, Object response) throws Exception {
-        SearchRequestBuilder searchRequestBuilder = EsClient.getClient().prepareSearch(NodeCompIndex.Index);
+        SearchRequestBuilder searchRequestBuilder = EsClient.INSTANCE.getClient().prepareSearch(NodeCompIndex.Index);
         searchRequestBuilder.setTypes(NodeCompIndex.Type_Record);
         searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
         searchRequestBuilder.setSize(100);

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

@@ -34,7 +34,7 @@ public class NodeMappingSearchWithTimeSlice extends AbstractLocalSyncWorker {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
 
-            SearchRequestBuilder searchRequestBuilder = EsClient.getClient().prepareSearch(NodeMappingIndex.Index);
+            SearchRequestBuilder searchRequestBuilder = EsClient.INSTANCE.getClient().prepareSearch(NodeMappingIndex.Index);
             searchRequestBuilder.setTypes(search.getSliceType());
             searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
             searchRequestBuilder.setQuery(QueryBuilders.rangeQuery(NodeMappingIndex.Time_Slice).gte(search.getStartTime()).lte(search.getEndTime()));

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

@@ -41,7 +41,7 @@ abstract class AbstractNodeRefAnalysis extends RecordAnalysisMember {
                     String front = segment.getApplicationCode();
                     dataJsonObj.addProperty(NodeRefIndex.Front, front);
 
-                    String behind = SpanPeersTools.getPeers(span);
+                    String behind = SpanPeersTools.INSTANCE.getPeers(span);
                     dataJsonObj.addProperty(NodeRefIndex.Behind, behind);
                     dataJsonObj.addProperty(NodeRefIndex.BehindIsRealCode, false);
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/noderef/persistence/NodeRefResSumGroupWithTimeSlice.java

@@ -35,7 +35,7 @@ public class NodeRefResSumGroupWithTimeSlice extends AbstractLocalSyncWorker {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
 
-            SearchRequestBuilder searchRequestBuilder = EsClient.getClient().prepareSearch(NodeRefResSumIndex.Index);
+            SearchRequestBuilder searchRequestBuilder = EsClient.INSTANCE.getClient().prepareSearch(NodeRefResSumIndex.Index);
             searchRequestBuilder.setTypes(search.getSliceType());
             searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
             searchRequestBuilder.setQuery(QueryBuilders.rangeQuery(NodeRefResSumIndex.Time_Slice).gte(search.getStartTime()).lte(search.getEndTime()));

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/noderef/persistence/NodeRefResSumSearchWithTimeSlice.java

@@ -36,7 +36,7 @@ public class NodeRefResSumSearchWithTimeSlice extends AbstractLocalSyncWorker {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
 
-            SearchRequestBuilder searchRequestBuilder = EsClient.getClient().prepareSearch(NodeRefResSumIndex.Index);
+            SearchRequestBuilder searchRequestBuilder = EsClient.INSTANCE.getClient().prepareSearch(NodeRefResSumIndex.Index);
             searchRequestBuilder.setTypes(search.getSliceType());
             searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
             searchRequestBuilder.setQuery(QueryBuilders.rangeQuery(NodeRefResSumIndex.Time_Slice).gte(search.getStartTime()).lte(search.getEndTime()));

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/noderef/persistence/NodeRefSearchWithTimeSlice.java

@@ -34,7 +34,7 @@ public class NodeRefSearchWithTimeSlice extends AbstractLocalSyncWorker {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
 
-            SearchRequestBuilder searchRequestBuilder = EsClient.getClient().prepareSearch(NodeRefIndex.Index);
+            SearchRequestBuilder searchRequestBuilder = EsClient.INSTANCE.getClient().prepareSearch(NodeRefIndex.Index);
             searchRequestBuilder.setTypes(search.getSliceType());
             searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
             searchRequestBuilder.setQuery(QueryBuilders.rangeQuery(NodeRefIndex.Time_Slice).gte(search.getStartTime()).lte(search.getEndTime()));

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

@@ -22,7 +22,7 @@ public class SegmentExceptionWithSegId extends AbstractLocalSyncWorker {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
 
-            GetResponse getResponse = EsClient.getClient().prepareGet(SegmentExceptionIndex.Index, SegmentExceptionIndex.Type_Record, search.segId).get();
+            GetResponse getResponse = EsClient.INSTANCE.getClient().prepareGet(SegmentExceptionIndex.Index, SegmentExceptionIndex.Type_Record, search.segId).get();
 
             JsonObject dataJson = new JsonObject();
             dataJson.addProperty(SegmentExceptionIndex.SegId, (String) getResponse.getSource().get(SegmentExceptionIndex.SegId));

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

@@ -41,7 +41,7 @@ public class SegmentTopSearchWithGlobalTraceId extends AbstractLocalSyncWorker {
     protected void onWork(Object request, Object response) throws Exception {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
-            Client client = EsClient.getClient();
+            Client client = EsClient.INSTANCE.getClient();
 
             String globalTraceData = client.prepareGet(GlobalTraceIndex.Index, GlobalTraceIndex.Type_Record, search.globalTraceId).get().getSourceAsString();
             JsonObject globalTraceObj = gson.fromJson(globalTraceData, JsonObject.class);

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

@@ -43,7 +43,7 @@ public class SegmentTopSearchWithTimeSlice extends AbstractLocalSyncWorker {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
 
-            SearchRequestBuilder searchRequestBuilder = EsClient.getClient().prepareSearch(SegmentCostIndex.Index);
+            SearchRequestBuilder searchRequestBuilder = EsClient.INSTANCE.getClient().prepareSearch(SegmentCostIndex.Index);
             searchRequestBuilder.setTypes(SegmentCostIndex.Type_Record);
             searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
             BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
@@ -86,7 +86,7 @@ public class SegmentTopSearchWithTimeSlice extends AbstractLocalSyncWorker {
                 topSegmentJson.addProperty(SegmentCostIndex.OperationName, (String) searchHit.getSource().get(SegmentCostIndex.OperationName));
                 topSegmentJson.addProperty(SegmentCostIndex.Cost, (Number) searchHit.getSource().get(SegmentCostIndex.Cost));
 
-                String segmentSource = EsClient.getClient().prepareGet(SegmentIndex.Index, SegmentIndex.Type_Record, segId).get().getSourceAsString();
+                String segmentSource = EsClient.INSTANCE.getClient().prepareGet(SegmentIndex.Index, SegmentIndex.Type_Record, segId).get().getSourceAsString();
                 Segment segment = SegmentDeserialize.INSTANCE.deserializeFromES(segmentSource);
                 List<DistributedTraceId> distributedTraceIdList = segment.getRelatedGlobalTraces();
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/span/persistence/SpanSearchWithId.java

@@ -29,7 +29,7 @@ public class SpanSearchWithId extends AbstractLocalSyncWorker {
     protected void onWork(Object request, Object response) throws Exception {
         if (request instanceof RequestEntity) {
             RequestEntity search = (RequestEntity) request;
-            GetResponse getResponse = EsClient.getClient().prepareGet(SegmentIndex.Index, SegmentIndex.Type_Record, search.segId).get();
+            GetResponse getResponse = EsClient.INSTANCE.getClient().prepareGet(SegmentIndex.Index, SegmentIndex.Type_Record, search.segId).get();
             Segment segment = SegmentDeserialize.INSTANCE.deserializeFromES(getResponse.getSourceAsString());
             List<Span> spanList = segment.getSpans();
 

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

@@ -29,20 +29,19 @@ public abstract class AbstractIndex {
     public static final String AGG_COLUMN = "aggId";
     public static final String Time_Slice = "timeSlice";
 
-    final public XContentBuilder createSettingBuilder() throws IOException {
-        XContentBuilder settingsBuilder = XContentFactory.jsonBuilder()
+    final XContentBuilder createSettingBuilder() throws IOException {
+        return XContentFactory.jsonBuilder()
                 .startObject()
                 .field("index.number_of_shards", 2)
                 .field("index.number_of_replicas", 0)
                 .endObject();
-        return settingsBuilder;
     }
 
     public abstract boolean isRecord();
 
     public abstract XContentBuilder createMappingBuilder() throws IOException;
 
-    final public void createIndex() {
+    final void createIndex() {
         // settings
         String settingSource = "";
 
@@ -59,7 +58,7 @@ public abstract class AbstractIndex {
             logger.error("create %s index mapping builder error", index());
         }
         Settings settings = Settings.builder().loadFromSource(settingSource).build();
-        IndicesAdminClient client = EsClient.getClient().admin().indices();
+        IndicesAdminClient client = EsClient.INSTANCE.getClient().admin().indices();
 
         if (isRecord()) {
             CreateIndexResponse response = client.prepareCreate(index()).setSettings(settings).addMapping(Type_Record, mappingBuilder).get();
@@ -74,8 +73,8 @@ public abstract class AbstractIndex {
         }
     }
 
-    final public boolean deleteIndex() {
-        IndicesAdminClient client = EsClient.getClient().admin().indices();
+    final boolean deleteIndex() {
+        IndicesAdminClient client = EsClient.INSTANCE.getClient().admin().indices();
         try {
             DeleteIndexResponse response = client.prepareDelete(index()).get();
             logger.info("delete %s index finished, isAcknowledged: %s", index(), response.isAcknowledged());

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

@@ -15,11 +15,12 @@ import java.net.UnknownHostException;
 /**
  * @author pengys5
  */
-public class EsClient {
+public enum EsClient {
+    INSTANCE;
 
-    private static Client client;
+    private Client client;
 
-    public static void boot() throws UnknownHostException {
+    public void boot() throws UnknownHostException {
         Settings settings = Settings.builder()
                 .put("cluster.name", "CollectorCluster")
                 .put("client.transport.sniff", true)
@@ -29,11 +30,11 @@ public class EsClient {
                 .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
     }
 
-    public static Client getClient() {
+    public Client getClient() {
         return client;
     }
 
-    public static void indexRefresh(String... indexName) {
+    public void indexRefresh(String... indexName) {
         Logger logger = LogManager.getFormatterLogger(EsClient.class);
         RefreshResponse response = client.admin().indices().refresh(new RefreshRequest(indexName)).actionGet();
         if (response.getShardFailures().length == response.getTotalShards()) {

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

 package com.a.eye.skywalking.collector.worker.storage;
 
-import com.a.eye.skywalking.collector.worker.globaltrace.GlobalTraceIndex;
-import com.a.eye.skywalking.collector.worker.node.NodeCompIndex;
-import com.a.eye.skywalking.collector.worker.node.NodeMappingIndex;
-import com.a.eye.skywalking.collector.worker.noderef.NodeRefIndex;
-import com.a.eye.skywalking.collector.worker.noderef.NodeRefResSumIndex;
-import com.a.eye.skywalking.collector.worker.segment.SegmentCostIndex;
-import com.a.eye.skywalking.collector.worker.segment.SegmentExceptionIndex;
-import com.a.eye.skywalking.collector.worker.segment.SegmentIndex;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.HashSet;
+import java.util.ServiceLoader;
+import java.util.Set;
 
 /**
  * @author pengys5
@@ -15,36 +13,23 @@ import com.a.eye.skywalking.collector.worker.segment.SegmentIndex;
 public enum IndexCreator {
     INSTANCE;
 
-    public void create() {
-        SegmentIndex segmentIndex = new SegmentIndex();
-        segmentIndex.deleteIndex();
-
-        GlobalTraceIndex globalTraceIndex = new GlobalTraceIndex();
-        globalTraceIndex.deleteIndex();
-        globalTraceIndex.createIndex();
-
-        SegmentCostIndex segmentCostIndex = new SegmentCostIndex();
-        segmentCostIndex.deleteIndex();
-        segmentCostIndex.createIndex();
-
-        SegmentExceptionIndex segmentExceptionIndex = new SegmentExceptionIndex();
-        segmentExceptionIndex.deleteIndex();
-        segmentExceptionIndex.createIndex();
+    private Logger logger = LogManager.getFormatterLogger(IndexCreator.class);
 
-        NodeCompIndex nodeCompIndex = new NodeCompIndex();
-        nodeCompIndex.deleteIndex();
-        nodeCompIndex.createIndex();
-
-        NodeMappingIndex nodeMappingIndex = new NodeMappingIndex();
-        nodeMappingIndex.deleteIndex();
-        nodeMappingIndex.createIndex();
-
-        NodeRefIndex nodeRefIndex = new NodeRefIndex();
-        nodeRefIndex.deleteIndex();
-        nodeRefIndex.createIndex();
+    public void create() {
+        Set<AbstractIndex> indexSet = loadIndex();
+        for (AbstractIndex index : indexSet) {
+            index.deleteIndex();
+            index.createIndex();
+        }
+    }
 
-        NodeRefResSumIndex nodeRefResSumIndex = new NodeRefResSumIndex();
-        nodeRefResSumIndex.deleteIndex();
-        nodeRefResSumIndex.createIndex();
+    private Set<AbstractIndex> loadIndex() {
+        Set<AbstractIndex> indexSet = new HashSet<>();
+        ServiceLoader<AbstractIndex> indexServiceLoader = ServiceLoader.load(AbstractIndex.class);
+        for (AbstractIndex index : indexServiceLoader) {
+            logger.info("index name: %s", index.index());
+            indexSet.add(index);
+        }
+        return indexSet;
     }
 }

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

@@ -11,7 +11,7 @@ import java.util.function.Consumer;
  */
 public class MergePersistenceData implements Iterable {
 
-    private Map<String, MergeData> persistenceData = new HashMap();
+    private Map<String, MergeData> persistenceData = new HashMap<>();
 
     public MergeData getElseCreate(String id) {
         if (!persistenceData.containsKey(id)) {
@@ -28,10 +28,6 @@ public class MergePersistenceData implements Iterable {
         persistenceData.clear();
     }
 
-    public boolean hasNext() {
-        return persistenceData.entrySet().iterator().hasNext();
-    }
-
     public MergeData pushOne() {
         MergeData one = persistenceData.entrySet().iterator().next().getValue();
         persistenceData.remove(one.getId());

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

@@ -10,7 +10,7 @@ import java.util.Spliterator;
  */
 public class MetricPersistenceData implements Iterable {
 
-    private Map<String, MetricData> persistenceData = new HashMap();
+    private Map<String, MetricData> persistenceData = new HashMap<>();
 
     public MetricData getElseCreate(String id) {
         if (!persistenceData.containsKey(id)) {

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

@@ -24,6 +24,7 @@ public class RecordData extends AbstractHashMessage {
                 this.aggId = this.aggId + Const.ID_SPLIT + ids[i];
             }
         }
+        record = new JsonObject();
     }
 
     public String getId() {

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

@@ -8,8 +8,10 @@ import com.a.eye.skywalking.trace.tag.Tags;
 /**
  * @author pengys5
  */
-public class SpanPeersTools {
-    public static String getPeers(Span span) {
+public enum SpanPeersTools {
+    INSTANCE;
+
+    public String getPeers(Span span) {
         if (StringUtil.isEmpty(Tags.PEERS.get(span))) {
             String host = Tags.PEER_HOST.get(span);
             int port = Tags.PEER_PORT.get(span);

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

-package com.a.eye.skywalking.collector.worker.tools;
-
-/**
- * @author pengys5
- */
-public class UrlTools {
-
-    private static final String HttpUrlHead = "http://";
-    private static final String HttpsUrlHead = "https://";
-    private static final String MotanUrlHead = "motan://";
-
-    public static String parse(String url, String component) {
-        if ("Tomcat".equals(component)) {
-            return parseTomcat(url);
-        } else if ("Motan".equals(component)) {
-            return parseMotan(url);
-        }
-        return null;
-    }
-
-    private static String parseTomcat(String url) {
-        if (url.startsWith(HttpUrlHead)) {
-            String suffix = url.substring(7, url.length());
-            String[] urlSplit = suffix.split("/");
-            return HttpUrlHead + urlSplit[0];
-        } else if (url.startsWith(HttpsUrlHead)) {
-            String suffix = url.substring(8, url.length());
-            String[] urlSplit = suffix.split("/");
-            return HttpsUrlHead + urlSplit[0];
-        } else if (url.contains(":")) {
-            return url.split("/")[0];
-        } else {
-            return url;
-        }
-    }
-
-    private static String parseMotan(String url) {
-        if (url.startsWith(MotanUrlHead)) {
-            String suffix = url.substring(8, url.length());
-            String[] urlSplit = suffix.split("/");
-            return MotanUrlHead + urlSplit[0];
-        } else {
-            return url;
-        }
-    }
-}

skywalking-collector/skywalking-collector-worker/src/main/resources/META-INF/services/com.a.eye.skywalking.collector.worker.storage.AbstractIndex

+com.a.eye.skywalking.collector.worker.segment.SegmentIndex
+com.a.eye.skywalking.collector.worker.segment.SegmentCostIndex
+com.a.eye.skywalking.collector.worker.segment.SegmentExceptionIndex
+
+com.a.eye.skywalking.collector.worker.globaltrace.GlobalTraceIndex
+
+com.a.eye.skywalking.collector.worker.node.NodeCompIndex
+com.a.eye.skywalking.collector.worker.node.NodeMappingIndex
+
+com.a.eye.skywalking.collector.worker.noderef.NodeRefIndex
+com.a.eye.skywalking.collector.worker.noderef.NodeRefResSumIndex
\ No newline at end of file

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

@@ -29,7 +29,7 @@ public class StartUpTestCase {
                 withFallback(ConfigFactory.load("application.conf"));
         ActorSystem system = ActorSystem.create("CollectorSystem", config);
 
-        EsClient.boot();
+        EsClient.INSTANCE.boot();
 
         TraceSegment dubboClientData = TraceSegmentBuilderFactory.INSTANCE.traceOf_Tomcat_DubboClient();
 

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/globaltrace/GlobalTraceSearchTestCase.java

@@ -28,7 +28,7 @@ public class GlobalTraceSearchTestCase {
     public void testSearchWithGlobalId() throws Exception {
         Client client = mock(Client.class);
         mockStatic(EsClient.class);
-        when(EsClient.getClient()).thenReturn(client);
+//        when(EsClient.INSTANCE.getClient()).thenReturn(client);
 
         String globalTraceId = "Global.1";
         String segment_1 = "SEGMENT.1";

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

@@ -7,6 +7,8 @@ import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.client.Client;
 import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.reflect.Whitebox;
 
 import static org.powermock.api.mockito.PowerMockito.*;
 
@@ -18,9 +20,10 @@ public class MockEsBulkClient {
     public IndexRequestBuilder indexRequestBuilder;
 
     public void createMock() {
-        Client client = mock(Client.class);
-        mockStatic(EsClient.class);
-        when(EsClient.getClient()).thenReturn(client);
+        Client client = PowerMockito.mock(Client.class);
+        EsClient esClient = PowerMockito.mock(EsClient.class);
+        Whitebox.setInternalState(EsClient.class, "INSTANCE", esClient);
+        when(esClient.getClient()).thenReturn(client);
 
         BulkRequestBuilder bulkRequestBuilder = mock(BulkRequestBuilder.class);
         when(client.prepareBulk()).thenReturn(bulkRequestBuilder);

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/node/persistence/NodeCompWithTimeSliceUseDB.java

@@ -12,7 +12,7 @@ import com.google.gson.JsonObject;
 public class NodeCompWithTimeSliceUseDB {
 
     public static void main(String[] args) throws Exception {
-        EsClient.boot();
+        EsClient.INSTANCE.boot();
 
         ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
         LocalWorkerContext localWorkerContext = new LocalWorkerContext();

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/node/persistence/NodeMappingSearchWithTimeSliceUseDB.java

@@ -13,7 +13,7 @@ import com.google.gson.JsonObject;
 public class NodeMappingSearchWithTimeSliceUseDB {
 
     public static void main(String[] args) throws Exception {
-        EsClient.boot();
+        EsClient.INSTANCE.boot();
 
         ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
         LocalWorkerContext localWorkerContext = new LocalWorkerContext();

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/noderef/persistence/NodeRefResSumGroupWithTimeSliceUseDB.java

@@ -13,7 +13,7 @@ import com.google.gson.JsonObject;
 public class NodeRefResSumGroupWithTimeSliceUseDB {
 
     public static void main(String[] args) throws Exception {
-        EsClient.boot();
+        EsClient.INSTANCE.boot();
 
         ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
         LocalWorkerContext localWorkerContext = new LocalWorkerContext();

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/noderef/persistence/NodeRefResSumSearchWithTimeSliceUseDB.java

@@ -13,7 +13,7 @@ import com.google.gson.JsonObject;
 public class NodeRefResSumSearchWithTimeSliceUseDB {
 
     public static void main(String[] args) throws Exception {
-        EsClient.boot();
+        EsClient.INSTANCE.boot();
 
         ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
         LocalWorkerContext localWorkerContext = new LocalWorkerContext();

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/noderef/persistence/NodeRefSearchWithTimeSliceUseDB.java

@@ -13,7 +13,7 @@ import com.google.gson.JsonObject;
 public class NodeRefSearchWithTimeSliceUseDB {
 
     public static void main(String[] args) throws Exception {
-        EsClient.boot();
+        EsClient.INSTANCE.boot();
 
         ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(null);
         LocalWorkerContext localWorkerContext = new LocalWorkerContext();

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

@@ -23,5 +23,9 @@ public class SegmentRealPost {
 
         String portalServiceSegmentAsString = mock.mockPortalServiceSegmentAsString();
         HttpClientTools.INSTANCE.post("http://localhost:7001/segments", portalServiceSegmentAsString);
+
+//        String specialSegmentAsString = mock.mockSpecialSegmentAsString();
+//        HttpClientTools.INSTANCE.post("http://localhost:7001/segments", specialSegmentAsString);
+
     }
 }

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

@@ -14,6 +14,7 @@ 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.elasticsearch.client.Client;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -31,7 +32,7 @@ import static org.powermock.api.mockito.PowerMockito.when;
  * @author pengys5
  */
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({EsClient.class, LogManager.class})
+@PrepareForTest({EsClient.class, Client.class, LogManager.class})
 @PowerMockIgnore({"javax.management.*"})
 public class SegmentCostSaveTestCase {
 

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/AbstractIndexTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.IOException;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({AbstractIndexTestCase.IndexTest.class, EsClient.class})
+@PowerMockIgnore({"javax.management.*"})
+public class AbstractIndexTestCase {
+
+    @Test
+    public void testCreateSettingBuilder() throws IOException {
+        IndexTest indexTest = new IndexTest();
+        Assert.assertEquals("{\"index.number_of_shards\":2,\"index.number_of_replicas\":0}", indexTest.createSettingBuilder().string());
+    }
+
+    class IndexTest extends AbstractIndex {
+        @Override
+        public boolean isRecord() {
+            return false;
+        }
+
+        @Override
+        public XContentBuilder createMappingBuilder() throws IOException {
+            XContentBuilder mappingBuilder = XContentFactory.jsonBuilder()
+                    .startObject()
+                    .startObject("properties")
+                    .startObject(Time_Slice)
+                    .field("type", "long")
+                    .field("index", "not_analyzed")
+                    .endObject()
+                    .endObject()
+                    .endObject();
+            return mappingBuilder;
+        }
+
+        @Override
+        public String index() {
+            return "Index_Test";
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/AbstractTimeSliceTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author pengys5
+ */
+public class AbstractTimeSliceTestCase {
+
+    @Test
+    public void test() {
+        TimeSlice timeSlice = new TimeSlice(1L, 2L, 3L, 4);
+        Assert.assertEquals(1L, timeSlice.getMinute());
+        Assert.assertEquals(2L, timeSlice.getHour());
+        Assert.assertEquals(3L, timeSlice.getDay());
+        Assert.assertEquals(4, timeSlice.getSecond());
+    }
+
+    class TimeSlice extends AbstractTimeSlice {
+        public TimeSlice(long minute, long hour, long day, int second) {
+            super(minute, hour, day, second);
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/IndexCreatorTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.junit.Assert;
+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 org.powermock.reflect.Whitebox;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.powermock.api.mockito.PowerMockito.*;
+
+/**
+ * @author pengys5
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({IndexCreator.class, IndexCreatorTestCase.TestIndex.class})
+@PowerMockIgnore({"javax.management.*"})
+public class IndexCreatorTestCase {
+
+    @Test
+    public void testLoadIndex() throws Exception {
+        IndexCreator indexCreator = spy(IndexCreator.INSTANCE);
+        Set<AbstractIndex> indexSet = Whitebox.invokeMethod(indexCreator, "loadIndex");
+        Assert.assertEquals(8, indexSet.size());
+
+        Set<String> indexName = new HashSet<>();
+        for (AbstractIndex index : indexSet) {
+            indexName.add(index.index());
+        }
+
+        Assert.assertEquals(true, indexName.contains("node_ref_idx"));
+        Assert.assertEquals(true, indexName.contains("node_ref_res_sum_idx"));
+        Assert.assertEquals(true, indexName.contains("global_trace_idx"));
+        Assert.assertEquals(true, indexName.contains("segment_cost_idx"));
+        Assert.assertEquals(true, indexName.contains("node_mapping_idx"));
+        Assert.assertEquals(true, indexName.contains("segment_idx"));
+        Assert.assertEquals(true, indexName.contains("segment_exp_idx"));
+        Assert.assertEquals(true, indexName.contains("node_comp_idx"));
+    }
+
+    @Test
+    public void testCreate() throws Exception {
+        TestIndex testIndex = mock(TestIndex.class);
+
+        IndexCreator indexCreator = mock(IndexCreator.class);
+        doCallRealMethod().when(indexCreator).create();
+
+        Set<AbstractIndex> indexSet = new HashSet<>();
+        indexSet.add(testIndex);
+
+        when(indexCreator, "loadIndex").thenReturn(indexSet);
+
+        indexCreator.create();
+        Mockito.verify(testIndex).createIndex();
+        Mockito.verify(testIndex).deleteIndex();
+    }
+
+    class TestIndex extends AbstractIndex {
+
+        @Override
+        public String index() {
+            return "TestIndex";
+        }
+
+        @Override
+        public boolean isRecord() {
+            return false;
+        }
+
+        @Override
+        public XContentBuilder createMappingBuilder() throws IOException {
+            return XContentFactory.jsonBuilder();
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/MergeDataTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public class MergeDataTestCase {
+
+    @Test
+    public void testConstruction() {
+        String id = "Test";
+        MergeData mergeData = new MergeData(id);
+
+        Assert.assertEquals(id, mergeData.getId());
+    }
+
+    @Test
+    public void testSetMergeData() {
+        String id = "Test";
+        MergeData mergeData = new MergeData(id);
+
+        mergeData.setMergeData("Column_1", "Value_1");
+        Assert.assertEquals("Value_1", mergeData.toMap().get("Column_1"));
+        mergeData.setMergeData("Column_1", "Value_1");
+        Assert.assertEquals("Value_1", mergeData.toMap().get("Column_1"));
+
+        mergeData.setMergeData("Column_1", "Value_2");
+        Assert.assertEquals("Value_2,Value_1", mergeData.toMap().get("Column_1"));
+
+        mergeData.setMergeData("Column_2", "Value_3");
+        Assert.assertEquals("Value_3", mergeData.toMap().get("Column_2"));
+    }
+
+    @Test
+    public void testMerge(){
+        String id = "Test";
+        MergeData mergeData_1 = new MergeData(id);
+        mergeData_1.setMergeData("Column_1", "Value_1");
+
+        MergeData mergeData_2 = new MergeData(id);
+        mergeData_2.setMergeData("Column_1", "Value_2");
+
+        mergeData_1.merge(mergeData_2);
+        Assert.assertEquals("Value_2,Value_1", mergeData_1.toMap().get("Column_1"));
+    }
+
+    @Test
+    public void testMergeMap(){
+        String id = "Test";
+        MergeData mergeData_1 = new MergeData(id);
+        mergeData_1.setMergeData("Column_1", "Value_1");
+
+        Map<String, Object> dbData = new HashMap<>();
+        dbData.put("Column_1", "Value_2");
+
+        mergeData_1.merge(dbData);
+        Assert.assertEquals("Value_2,Value_1", mergeData_1.toMap().get("Column_1"));
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/MergePersistenceDataTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public class MergePersistenceDataTestCase {
+
+    @Test
+    public void testGetElseCreate() {
+        MergePersistenceData persistenceData = new MergePersistenceData();
+        MergeData mergeData = persistenceData.getElseCreate("test");
+        Assert.assertEquals("test", mergeData.getId());
+    }
+
+    @Test
+    public void testSize() {
+        MergePersistenceData persistenceData = new MergePersistenceData();
+        persistenceData.getElseCreate("test_1");
+        Assert.assertEquals(1, persistenceData.size());
+        persistenceData.getElseCreate("test_1");
+        Assert.assertEquals(1, persistenceData.size());
+        persistenceData.getElseCreate("test_2");
+        Assert.assertEquals(2, persistenceData.size());
+    }
+
+    @Test
+    public void testClear() {
+        MergePersistenceData persistenceData = new MergePersistenceData();
+        persistenceData.getElseCreate("test_1");
+        Assert.assertEquals(1, persistenceData.size());
+        persistenceData.clear();
+        Assert.assertEquals(0, persistenceData.size());
+    }
+
+    @Test
+    public void testPushOne() {
+        MergePersistenceData persistenceData = new MergePersistenceData();
+        persistenceData.getElseCreate("test_1");
+        persistenceData.getElseCreate("test_2");
+        persistenceData.getElseCreate("test_3");
+
+        Assert.assertEquals(3, persistenceData.size());
+        MergeData mergeData = persistenceData.pushOne();
+        Assert.assertEquals("test_3", mergeData.getId());
+        Assert.assertEquals(2, persistenceData.size());
+
+        mergeData = persistenceData.pushOne();
+        Assert.assertEquals("test_2", mergeData.getId());
+        Assert.assertEquals(1, persistenceData.size());
+
+        mergeData = persistenceData.pushOne();
+        Assert.assertEquals("test_1", mergeData.getId());
+        Assert.assertEquals(0, persistenceData.size());
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testForEach() {
+        MergePersistenceData persistenceData = new MergePersistenceData();
+        persistenceData.forEach(c -> System.out.println(c));
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSpliterator() {
+        MergePersistenceData persistenceData = new MergePersistenceData();
+        persistenceData.spliterator();
+    }
+
+    @Test
+    public void testIterator() {
+        MergePersistenceData persistenceData = new MergePersistenceData();
+        persistenceData.getElseCreate("test_1");
+        persistenceData.getElseCreate("test_2");
+        persistenceData.getElseCreate("test_3");
+
+        Iterator<Map.Entry<String, MergeData>> iterator = persistenceData.iterator();
+        Assert.assertEquals("test_3", iterator.next().getKey());
+        Assert.assertEquals("test_2", iterator.next().getKey());
+        Assert.assertEquals("test_1", iterator.next().getKey());
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/MetricDataTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import com.a.eye.skywalking.collector.worker.Const;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public class MetricDataTestCase {
+
+    @Test
+    public void testConstruction() {
+        String id_1 = "2016" + Const.ID_SPLIT + "B";
+        MetricData metricData_1 = new MetricData(id_1);
+
+        Assert.assertEquals(id_1, metricData_1.getId());
+        Assert.assertEquals(2016L, metricData_1.toMap().get(AbstractIndex.Time_Slice));
+        Assert.assertEquals("B", metricData_1.toMap().get(AbstractIndex.AGG_COLUMN));
+
+        String id_2 = "2017" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+        MetricData metricData_2 = new MetricData(id_2);
+
+        Assert.assertEquals(id_2, metricData_2.getId());
+        Assert.assertEquals(2017L, metricData_2.toMap().get(AbstractIndex.Time_Slice));
+        Assert.assertEquals("B" + Const.ID_SPLIT + "C", metricData_2.toMap().get(AbstractIndex.AGG_COLUMN));
+    }
+
+    @Test
+    public void testSetMetric() {
+        String id_1 = "2016" + Const.ID_SPLIT + "B";
+        MetricData metricData = new MetricData(id_1);
+
+        metricData.setMetric("Column", 10L);
+        Assert.assertEquals(10L, metricData.toMap().get("Column"));
+
+        metricData.setMetric("Column", 10L);
+        Assert.assertEquals(20L, metricData.toMap().get("Column"));
+    }
+
+    @Test
+    public void testMerge() {
+        String id_1 = "2016" + Const.ID_SPLIT + "B";
+        MetricData metricData_1 = new MetricData(id_1);
+        metricData_1.setMetric("Column", 10L);
+
+        MetricData metricData_2 = new MetricData(id_1);
+        metricData_2.setMetric("Column", 10L);
+
+        metricData_1.merge(metricData_2);
+        Assert.assertEquals(20L, metricData_1.toMap().get("Column"));
+    }
+
+    @Test
+    public void testMergeMapData() {
+        String id_1 = "2016" + Const.ID_SPLIT + "B";
+        MetricData metricData_1 = new MetricData(id_1);
+        metricData_1.setMetric("Column", 10L);
+
+        Map<String, Object> dbData = new HashMap<>();
+        dbData.put("Column", 10L);
+
+        metricData_1.merge(dbData);
+        Assert.assertEquals(20L, metricData_1.toMap().get("Column"));
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/MetricPersistenceDataTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import com.a.eye.skywalking.collector.worker.Const;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public class MetricPersistenceDataTestCase {
+
+    @Test
+    public void testGetElseCreate() {
+        String id = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+
+        MetricPersistenceData metricPersistenceData = new MetricPersistenceData();
+        MetricData metricData = metricPersistenceData.getElseCreate(id);
+        metricData.setMetric("Column_1", 10L);
+        Assert.assertEquals(id, metricData.getId());
+
+        MetricData metricData1 = metricPersistenceData.getElseCreate(id);
+        Assert.assertEquals(10L, metricData1.toMap().get("Column_1"));
+    }
+
+    @Test
+    public void testSize() {
+        String id = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+
+        MetricPersistenceData metricPersistenceData = new MetricPersistenceData();
+        metricPersistenceData.getElseCreate(id);
+
+        Assert.assertEquals(1, metricPersistenceData.size());
+        String id_1 = "2016" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+        metricPersistenceData.getElseCreate(id_1);
+        Assert.assertEquals(2, metricPersistenceData.size());
+
+        metricPersistenceData.clear();
+        Assert.assertEquals(0, metricPersistenceData.size());
+    }
+
+    @Test
+    public void testPushOne() {
+        String id_1 = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+        String id_2 = "2016" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+
+        MetricPersistenceData metricPersistenceData = new MetricPersistenceData();
+        metricPersistenceData.getElseCreate(id_1);
+        metricPersistenceData.getElseCreate(id_2);
+
+        MetricData metricData_2 = metricPersistenceData.pushOne();
+        Assert.assertEquals(id_2, metricData_2.getId());
+
+        MetricData metricData_1 = metricPersistenceData.pushOne();
+        Assert.assertEquals(id_1, metricData_1.getId());
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSpliterator() {
+        MetricPersistenceData metricPersistenceData = new MetricPersistenceData();
+        metricPersistenceData.spliterator();
+    }
+
+    @Test
+    public void testIterator() {
+        String id_1 = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+        String id_2 = "2016" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+
+        MetricPersistenceData metricPersistenceData = new MetricPersistenceData();
+        metricPersistenceData.getElseCreate(id_1);
+        metricPersistenceData.getElseCreate(id_2);
+
+        Iterator<Map.Entry<String, MetricData>> iterator = metricPersistenceData.iterator();
+        Assert.assertEquals(id_2, iterator.next().getKey());
+        Assert.assertEquals(id_1, iterator.next().getKey());
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/RecordDataTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import com.a.eye.skywalking.collector.worker.Const;
+import com.google.gson.JsonObject;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author pengys5
+ */
+public class RecordDataTestCase {
+
+    @Test
+    public void testConstruction() {
+        String id_1 = "2017" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+        RecordData recordData = new RecordData(id_1);
+
+        Assert.assertEquals(id_1, recordData.getId());
+        Assert.assertEquals("B" + Const.ID_SPLIT + "C", recordData.getRecord().get("aggId").getAsString());
+    }
+
+    @Test
+    public void testSetRecord() {
+        String id_1 = "2017" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+        RecordData recordData = new RecordData(id_1);
+
+        JsonObject record = new JsonObject();
+        record.addProperty("Column", "Value");
+        recordData.setRecord(record);
+
+        Assert.assertEquals(id_1, recordData.getId());
+        Assert.assertEquals("B" + Const.ID_SPLIT + "C", recordData.getRecord().get("aggId").getAsString());
+        Assert.assertEquals("Value", recordData.getRecord().get("Column").getAsString());
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/storage/RecordPersistenceDataTestCase.java

+package com.a.eye.skywalking.collector.worker.storage;
+
+import com.a.eye.skywalking.collector.worker.Const;
+import com.google.gson.JsonObject;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public class RecordPersistenceDataTestCase {
+
+    @Test
+    public void testGetElseCreate() {
+        String id = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+
+        JsonObject record = new JsonObject();
+        record.addProperty("Column_1", "Value_1");
+        RecordPersistenceData recordPersistenceData = new RecordPersistenceData();
+        RecordData recordData = recordPersistenceData.getElseCreate(id);
+        recordData.setRecord(record);
+
+        Assert.assertEquals(id, recordData.getId());
+
+        RecordData recordData1 = recordPersistenceData.getElseCreate(id);
+        Assert.assertEquals("Value_1", recordData1.getRecord().get("Column_1").getAsString());
+    }
+
+    @Test
+    public void testClear() {
+        String id_1 = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+        String id_2 = "2016" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+        RecordPersistenceData recordPersistenceData = new RecordPersistenceData();
+        recordPersistenceData.getElseCreate(id_1);
+        Assert.assertEquals(1, recordPersistenceData.size());
+        recordPersistenceData.getElseCreate(id_2);
+        Assert.assertEquals(2, recordPersistenceData.size());
+
+        Assert.assertEquals(true, recordPersistenceData.hasNext());
+
+        recordPersistenceData.clear();
+        Assert.assertEquals(0, recordPersistenceData.size());
+
+        Assert.assertEquals(false, recordPersistenceData.hasNext());
+    }
+
+    @Test
+    public void testPushOne() {
+        String id_1 = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+        String id_2 = "2016" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+        RecordPersistenceData recordPersistenceData = new RecordPersistenceData();
+        JsonObject obj_1 = new JsonObject();
+        obj_1.addProperty("Column_1", "Value_1");
+        recordPersistenceData.getElseCreate(id_1).setRecord(obj_1);
+
+        JsonObject obj_2 = new JsonObject();
+        obj_2.addProperty("Column_2", "Value_2");
+        recordPersistenceData.getElseCreate(id_2).setRecord(obj_2);
+
+        RecordData recordData_2 = recordPersistenceData.pushOne();
+        Assert.assertEquals("Value_2", recordData_2.getRecord().get("Column_2").getAsString());
+
+        RecordData recordData_1 = recordPersistenceData.pushOne();
+        Assert.assertEquals("Value_1", recordData_1.getRecord().get("Column_1").getAsString());
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testForEach() {
+        RecordPersistenceData recordPersistenceData = new RecordPersistenceData();
+        recordPersistenceData.forEach(r -> System.out.println(r));
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testSpliterator() {
+        RecordPersistenceData recordPersistenceData = new RecordPersistenceData();
+        recordPersistenceData.spliterator();
+    }
+
+    @Test
+    public void testIterator(){
+        String id_1 = "2016" + Const.ID_SPLIT + "A" + Const.ID_SPLIT + "B";
+        String id_2 = "2016" + Const.ID_SPLIT + "B" + Const.ID_SPLIT + "C";
+        RecordPersistenceData recordPersistenceData = new RecordPersistenceData();
+        JsonObject obj_1 = new JsonObject();
+        obj_1.addProperty("Column_1", "Value_1");
+        recordPersistenceData.getElseCreate(id_1).setRecord(obj_1);
+
+        JsonObject obj_2 = new JsonObject();
+        obj_2.addProperty("Column_2", "Value_2");
+        recordPersistenceData.getElseCreate(id_2).setRecord(obj_2);
+
+        Iterator<Map.Entry<String, RecordData>> iterator = recordPersistenceData.iterator();
+        Assert.assertEquals("Value_2", iterator.next().getValue().getRecord().get("Column_2").getAsString());
+        Assert.assertEquals("Value_1", iterator.next().getValue().getRecord().get("Column_1").getAsString());
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/tools/DateToolsTestCase.java

 package com.a.eye.skywalking.collector.worker.tools;
 
+import com.a.eye.skywalking.collector.worker.storage.IndexCreator;
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.util.TimeZone;
 
+import static org.powermock.api.mockito.PowerMockito.spy;
+
 /**
  * @author pengys5
  */

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/tools/SpanPeersToolsTestCase.java

+package com.a.eye.skywalking.collector.worker.tools;
+
+import com.a.eye.skywalking.trace.Span;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.mockito.Mockito.*;
+
+/**
+ * @author pengys5
+ */
+public class SpanPeersToolsTestCase {
+
+    @Test
+    public void testNotEmptyPeers() {
+        Span span = mock(Span.class);
+        when(span.getStrTag("peers")).thenReturn("Test");
+
+        String peers = SpanPeersTools.INSTANCE.getPeers(span);
+        Assert.assertEquals("[Test]", peers);
+    }
+
+    @Test
+    public void testEmptyPeers() {
+        Span span = mock(Span.class);
+        when(span.getStrTag("peers")).thenReturn(null);
+        when(span.getStrTag("peer.host")).thenReturn("localhost");
+        when(span.getStrTag("peer.port")).thenReturn("8080");
+
+        String peers = SpanPeersTools.INSTANCE.getPeers(span);
+        Assert.assertEquals("[localhost:0]", peers);
+    }
+}

skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/tools/UrlToolsTestCase.java

-package com.a.eye.skywalking.collector.worker.tools;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-/**
- * @author pengys5
- */
-public class UrlToolsTestCase {
-
-//    @Test
-    public void testParseTomcat() {
-        String peers =
-                UrlTools.parse("http://172.0.0.1:8080/Web/GetUser", "Tomcat");
-        Assert.assertEquals(peers, "http://172.0.0.1:8080");
-
-        peers = UrlTools.parse("https://172.0.0.1:8080/Web/GetUser", "Tomcat");
-        Assert.assertEquals(peers, "https://172.0.0.1:8080");
-
-        peers = UrlTools.parse("172.0.0.1:8080/Web/GetUser", "Tomcat");
-        Assert.assertEquals(peers, "172.0.0.1:8080");
-
-        peers = UrlTools.parse("http172.0.0.18080/Web/GetUser", "Tomcat");
-        Assert.assertEquals(peers, "http172.0.0.18080/Web/GetUser");
-    }
-
-//    @Test
-    public void testParseMotan() {
-        String peers =
-                UrlTools.parse("motan://10.20.3.15:3000/com.a.eye.skywalking.demo.services.GetUserService.findUser(String, String)", "Motan");
-        Assert.assertEquals(peers, "motan://10.20.3.15:3000");
-    }
-
-}

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

@@ -3,18 +3,10 @@
     "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,
+        "si": 0,
+        "ps": -1,
         "st": 1491277162141,
         "et": 1491277162144,
         "on": "Jedis/getClient",
@@ -30,153 +22,6 @@
           "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",