collector worker finish, but not test

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMember.java

@@ -11,16 +11,26 @@ import java.util.List;
  */
 public abstract class AbstractMember<T> {
 
+    private MemberSystem memberSystem;
+
     private ActorRef actorRef;
 
+    public MemberSystem memberContext() {
+        return memberSystem;
+    }
+
     public ActorRef getSelf() {
         return actorRef;
     }
 
-    public void creatorRef(ActorRef actorRef) {
+    public AbstractMember(MemberSystem memberSystem, ActorRef actorRef) {
+        this.memberSystem = memberSystem;
         this.actorRef = actorRef;
     }
 
+
+    public abstract void preStart() throws Throwable;
+
     /**
      * Receive the message to analyse.
      *

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMemberProvider.java

@@ -2,19 +2,23 @@ package com.a.eye.skywalking.collector.actor;
 
 import akka.actor.ActorRef;
 
+import java.lang.reflect.Constructor;
+
 /**
  * @author pengys5
  */
 public abstract class AbstractMemberProvider {
     public abstract Class memberClass();
 
-    public void createWorker(MemberSystem system, ActorRef actorRef) {
+    public void createWorker(MemberSystem system, ActorRef actorRef) throws Exception {
         if (memberClass() == null) {
             throw new IllegalArgumentException("cannot createInstance() with nothing obtained from memberClass()");
         }
 
-        AbstractMember member = system.memberOf(memberClass(), roleName());
-        member.creatorRef(actorRef);
+        Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{MemberSystem.class, ActorRef.class});
+        memberConstructor.setAccessible(true);
+        AbstractMember member = (AbstractMember) memberConstructor.newInstance(system, actorRef);
+        system.memberOf(member, roleName());
     }
 
     /**

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorker.java

@@ -95,7 +95,7 @@ public abstract class AbstractWorker<T> extends UntypedActor {
         }
     }
 
-    public MemberSystem getMemberContext() {
+    public MemberSystem memberContext() {
         return memberSystem;
     }
 }

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/MemberSystem.java

@@ -10,17 +10,8 @@ public class MemberSystem {
 
     private Map<String, AbstractMember> memberMap = new HashMap();
 
-    public AbstractMember memberOf(Class clazz, String role) {
-        try {
-            AbstractMember member = (AbstractMember) clazz.newInstance();
-            memberMap.put(role, member);
-            return member;
-        } catch (InstantiationException e) {
-            e.printStackTrace();
-        } catch (IllegalAccessException e) {
-            e.printStackTrace();
-        }
-        return null;
+    public void memberOf(AbstractMember member, String role) {
+        memberMap.put(role, member);
     }
 
     public AbstractMember memberFor(String role) {

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkersCreator.java

@@ -19,8 +19,8 @@ public enum WorkersCreator {
      * @param system is create by akka {@link ActorSystem}
      */
     public void boot(ActorSystem system) {
-        ServiceLoader<AbstractClusterWorkerProvider> clusterServiceLoader = ServiceLoader.load(AbstractClusterWorkerProvider.class);
-        for (AbstractClusterWorkerProvider provider : clusterServiceLoader) {
+        ServiceLoader<AbstractWorkerProvider> clusterServiceLoader = ServiceLoader.load(AbstractWorkerProvider.class);
+        for (AbstractWorkerProvider provider : clusterServiceLoader) {
             provider.createWorker(system);
         }
     }

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/selector/LocalSelector.java

@@ -0,0 +1,36 @@
+package com.a.eye.skywalking.collector.actor.selector;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorker;
+import com.a.eye.skywalking.collector.actor.WorkerRef;
+
+import java.util.List;
+
+/**
+ * The <code>LocalSelector</code> is a simple implementation of {@link WorkerSelector}.
+ * It choose {@link WorkerRef} nearly random, by round-robin.
+ *
+ * @author wusheng
+ */
+public enum LocalSelector implements WorkerSelector<Object> {
+    INSTANCE;
+
+    /**
+     * A simple round variable.
+     */
+    private int index = 0;
+
+    /**
+     * Use round-robin to select {@link WorkerRef}.
+     *
+     * @param members given {@link WorkerRef} list, which size is greater than 0;
+     * @param message the {@link AbstractWorker} is going to send.
+     * @return the selected {@link WorkerRef}
+     */
+    @Override
+    public WorkerRef select(List<WorkerRef> members, Object message) {
+        int size = members.size();
+        index++;
+        int selectIndex = Math.abs(index) % size;
+        return members.get(selectIndex);
+    }
+}

skywalking-collector/skywalking-collector-cluster/src/test/resources/META-INF/services/com.a.eye.skywalking.collector.actor.AbstractMemberProvider

@@ -1 +0,0 @@
-com.a.eye.skywalking.collector.actor.SpiTestWorkerFactory

skywalking-collector/skywalking-collector-worker/pom.xml

@@ -23,5 +23,10 @@
             <artifactId>gson</artifactId>
             <version>2.8.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>transport</artifactId>
+            <version>5.2.2</version>
+        </dependency>
     </dependencies>
 </project>

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

@@ -1,40 +0,0 @@
-package com.a.eye.skywalking.collector.worker;
-
-/**
- * @author pengys5
- */
-public class Metric {
-    private String timeSlice;
-    private String metricName;
-    private Long metricValue;
-
-    public Metric(String timeSlice, String metricName, Long metricValue) {
-        this.timeSlice = timeSlice;
-        this.metricName = metricName;
-        this.metricValue = metricValue;
-    }
-
-    public String getTimeSlice() {
-        return timeSlice;
-    }
-
-    public void setTimeSlice(String timeSlice) {
-        this.timeSlice = timeSlice;
-    }
-
-    public String getMetricName() {
-        return metricName;
-    }
-
-    public void setMetricName(String metricName) {
-        this.metricName = metricName;
-    }
-
-    public Long getMetricValue() {
-        return metricValue;
-    }
-
-    public void setMetricValue(Long metricValue) {
-        this.metricValue = metricValue;
-    }
-}

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

@@ -1,22 +0,0 @@
-package com.a.eye.skywalking.collector.worker;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * @author pengys5
- */
-public class MetricCollection {
-    private Map<String, Metric> metricMap = new HashMap();
-
-    public void put(String timeSlice, String name, Long value) {
-        String timeSliceName = name + timeSlice;
-
-        if (metricMap.containsKey(timeSliceName)) {
-            Long metric = metricMap.get(timeSliceName).getMetricValue();
-            metricMap.get(timeSliceName).setMetricValue(metric + value);
-        } else {
-            metricMap.put(timeSliceName, new Metric(timeSlice, name, value));
-        }
-    }
-}

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

@@ -0,0 +1,7 @@
+package com.a.eye.skywalking.collector.worker;
+
+/**
+ * @author pengys5
+ */
+public class PersistenceCommand {
+}

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

@@ -0,0 +1,75 @@
+package com.a.eye.skywalking.collector.worker;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorker;
+import com.a.eye.skywalking.collector.worker.PersistenceCommand;
+import com.a.eye.skywalking.collector.worker.tools.EsClient;
+import com.google.gson.JsonObject;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author pengys5
+ */
+public abstract class PersistenceWorker<T> extends AbstractWorker<T> {
+
+    private long lastPersistenceTimestamp = 0;
+
+    private Map<String, JsonObject> persistenceData = new HashMap();
+
+    public abstract String esIndex();
+
+    public abstract String esType();
+
+    public void putData(String id, JsonObject data) {
+        persistenceData.put(id, data);
+        if (persistenceData.size() >= 1000) {
+            persistence(true);
+        }
+    }
+
+    public boolean containsId(String id) {
+        return persistenceData.containsKey(id);
+    }
+
+    public JsonObject getData(String id) {
+        return persistenceData.get(id);
+    }
+
+    public abstract void analyse(Object message) throws Throwable;
+
+    @Override
+    public void receive(Object message) throws Throwable {
+        if (message instanceof PersistenceCommand) {
+            persistence(false);
+        } else {
+            analyse(message);
+        }
+    }
+
+    private void persistence(boolean dataFull) {
+        long now = System.currentTimeMillis();
+        if (now - lastPersistenceTimestamp > 5000 || dataFull) {
+            boolean success = saveToEs();
+            if (success) {
+                persistenceData.clear();
+                lastPersistenceTimestamp = now;
+            }
+        }
+    }
+
+    private boolean saveToEs() {
+        BulkRequestBuilder bulkRequest = EsClient.client().prepareBulk();
+
+        for (Map.Entry<String, JsonObject> entry : persistenceData.entrySet()) {
+            String id = entry.getKey();
+            JsonObject data = entry.getValue();
+            bulkRequest.add(EsClient.client().prepareIndex(esIndex(), esType(), id).setSource(data.toString()));
+        }
+
+        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
+        return !bulkResponse.hasFailures();
+    }
+}

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

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker;
-
-import com.google.gson.JsonObject;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * @author pengys5
- */
-public class RecordCollection {
-
-    private Map<String, JsonObject> recordMap = new HashMap();
-
-    public void put(String timeSlice, String primaryKey, JsonObject valueObj) {
-        recordMap.put(timeSlice + primaryKey, valueObj);
-    }
-}

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

@@ -1,16 +0,0 @@
-package com.a.eye.skywalking.collector.worker;
-
-/**
- * @author pengys5
- */
-public abstract class TimeSliceMessage {
-    private final String timeSlice;
-
-    public TimeSliceMessage(String timeSlice) {
-        this.timeSlice = timeSlice;
-    }
-
-    public String getTimeSlice() {
-        return timeSlice;
-    }
-}

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

@@ -7,4 +7,15 @@ import com.a.eye.skywalking.collector.cluster.ClusterConfig;
  */
 public class WorkerConfig extends ClusterConfig {
 
+    public static class WorkerNum {
+        public static int TraceSegmentReceiver_Num = 1;
+
+        public static int DAGNodePersistence_Num = 5;
+        public static int DAGNodeRefPersistence_Num = 5;
+        public static int NodeInstancePersistence_Num = 5;
+        public static int ResponseCostPersistence_Num = 5;
+        public static int ResponseSummaryPersistence_Num = 5;
+        public static int TraceSegmentRecordPersistence_Num = 5;
+    }
+
 }

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

@@ -0,0 +1,108 @@
+package com.a.eye.skywalking.collector.worker.application;
+
+import akka.actor.ActorRef;
+import com.a.eye.skywalking.collector.actor.AbstractMember;
+import com.a.eye.skywalking.collector.actor.AbstractMemberProvider;
+import com.a.eye.skywalking.collector.actor.MemberSystem;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.application.metric.TraceSegmentRecordMember;
+import com.a.eye.skywalking.collector.worker.application.persistence.DAGNodePersistence;
+import com.a.eye.skywalking.collector.worker.application.persistence.NodeInstancePersistence;
+import com.a.eye.skywalking.collector.worker.application.persistence.ResponseCostPersistence;
+import com.a.eye.skywalking.collector.worker.application.persistence.ResponseSummaryPersistence;
+import com.a.eye.skywalking.trace.Span;
+import com.a.eye.skywalking.trace.TraceSegment;
+import com.a.eye.skywalking.trace.tag.Tags;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationMember extends AbstractMember {
+
+    public ApplicationMember(MemberSystem memberSystem, ActorRef actorRef) {
+        super(memberSystem, actorRef);
+    }
+
+    @Override
+    public void preStart() throws Throwable {
+        TraceSegmentRecordMember.Factory factory = new TraceSegmentRecordMember.Factory();
+        factory.createWorker(memberContext(), getSelf());
+    }
+
+    @Override
+    public void receive(Object message) throws Throwable {
+        if (message instanceof TraceSegment) {
+            TraceSegment traceSegment = (TraceSegment) message;
+            AbstractMember discoverMember = memberContext().memberFor(TraceSegmentRecordMember.class.getSimpleName());
+            discoverMember.receive(traceSegment);
+
+            sendToDAGNodePersistence(traceSegment);
+            sendToNodeInstancePersistence(traceSegment);
+            sendToResponseCostPersistence(traceSegment);
+            sendToResponseSummaryPersistence(traceSegment);
+        }
+    }
+
+    public static class Factory extends AbstractMemberProvider {
+        @Override
+        public Class memberClass() {
+            return ApplicationMember.class;
+        }
+    }
+
+    private void sendToDAGNodePersistence(TraceSegment traceSegment) throws Throwable {
+        String code = traceSegment.getApplicationCode();
+
+        String component = null;
+        String layer = null;
+        for (Span span : traceSegment.getSpans()) {
+            if (span.getParentSpanId() == -1) {
+                component = Tags.COMPONENT.get(span);
+                layer = Tags.SPAN_LAYER.get(span);
+            }
+        }
+
+        DAGNodePersistence.Metric node = new DAGNodePersistence.Metric(code, component, layer);
+        tell(new NodeInstancePersistence.Factory(), RollingSelector.INSTANCE, node);
+    }
+
+    private void sendToNodeInstancePersistence(TraceSegment traceSegment) throws Throwable {
+        String code = traceSegment.getPrimaryRef().getApplicationCode();
+        String address = traceSegment.getPrimaryRef().getPeerHost();
+
+        NodeInstancePersistence.Metric property = new NodeInstancePersistence.Metric(code, address);
+        tell(new NodeInstancePersistence.Factory(), RollingSelector.INSTANCE, property);
+    }
+
+    private void sendToResponseCostPersistence(TraceSegment traceSegment) throws Throwable {
+        String code = traceSegment.getApplicationCode();
+        long startTime = -1;
+        long endTime = -1;
+        boolean isError = false;
+
+        for (Span span : traceSegment.getSpans()) {
+            if (span.getParentSpanId() == -1) {
+                startTime = span.getStartTime();
+                endTime = span.getEndTime();
+                isError = Tags.ERROR.get(span);
+            }
+        }
+
+        ResponseCostPersistence.Metric cost = new ResponseCostPersistence.Metric(code, isError, startTime, endTime);
+        tell(new ResponseCostPersistence.Factory(), RollingSelector.INSTANCE, cost);
+    }
+
+    private void sendToResponseSummaryPersistence(TraceSegment traceSegment) throws Throwable {
+        String code = traceSegment.getApplicationCode();
+        boolean isError = false;
+
+        for (Span span : traceSegment.getSpans()) {
+            if (span.getParentSpanId() == -1) {
+                isError = Tags.ERROR.get(span);
+            }
+        }
+
+        ResponseSummaryPersistence.Metric summary = new ResponseSummaryPersistence.Metric(code, isError);
+        tell(new ResponseSummaryPersistence.Factory(), RollingSelector.INSTANCE, summary);
+    }
+}

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

@@ -1,30 +0,0 @@
-package com.a.eye.skywalking.collector.worker.application;
-
-import com.a.eye.skywalking.collector.actor.AbstractMember;
-import com.a.eye.skywalking.collector.actor.AbstractWorker;
-import com.a.eye.skywalking.collector.worker.application.member.ApplicationDiscoverFactory;
-import com.a.eye.skywalking.collector.worker.application.member.ApplicationDiscoverMember;
-import com.a.eye.skywalking.trace.TraceSegment;
-
-/**
- * @author pengys5
- */
-public class ApplicationWorker extends AbstractWorker {
-
-    @Override
-    public void preStart() throws Exception {
-        ApplicationDiscoverFactory factory = new ApplicationDiscoverFactory();
-        factory.createWorker(getMemberContext(), getSelf());
-
-        super.preStart();
-    }
-
-    @Override
-    public void receive(Object message) throws Throwable {
-        if (message instanceof TraceSegment) {
-            TraceSegment traceSegment = (TraceSegment) message;
-            AbstractMember discoverMember = getMemberContext().memberFor(ApplicationDiscoverMember.class.getSimpleName());
-            discoverMember.receive(traceSegment);
-        }
-    }
-}

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

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker.application;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-
-/**
- * @author pengys5
- */
-public class ApplicationWorkerFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return ApplicationWorker.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/member/ApplicationDiscoverFactory.java

@@ -1,14 +0,0 @@
-package com.a.eye.skywalking.collector.worker.application.member;
-
-import com.a.eye.skywalking.collector.actor.AbstractMemberProvider;
-
-/**
- * @author pengys5
- */
-public class ApplicationDiscoverFactory extends AbstractMemberProvider {
-
-    @Override
-    public Class memberClass() {
-        return ApplicationDiscoverMember.class;
-    }
-}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/member/ApplicationDiscoverMember.java

@@ -1,31 +0,0 @@
-package com.a.eye.skywalking.collector.worker.application.member;
-
-
-import com.a.eye.skywalking.collector.actor.AbstractMember;
-import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
-import com.a.eye.skywalking.collector.worker.application.persistence.ApplicationMessage;
-import com.a.eye.skywalking.collector.worker.application.persistence.ApplicationPersistenceFactory;
-import com.a.eye.skywalking.trace.TraceSegment;
-import com.a.eye.skywalking.trace.tag.Tags;
-
-/**
- * @author pengys5
- */
-public class ApplicationDiscoverMember extends AbstractMember {
-
-    @Override
-    public void receive(Object message) throws Throwable {
-        if (message instanceof TraceSegment) {
-            TraceSegment traceSegment = (TraceSegment) message;
-            String code = traceSegment.getApplicationCode();
-            String component = Tags.COMPONENT.get(traceSegment.getSpans().get(0));
-            String host = Tags.PEER_HOST.get(traceSegment.getSpans().get(0));
-            int port = Tags.PEER_PORT.get(traceSegment.getSpans().get(0));
-            String layer = Tags.SPAN_LAYER.get(traceSegment.getSpans().get(0));
-
-            ApplicationMessage applicationMessage = new ApplicationMessage(code, component, host, layer);
-            tell(new ApplicationPersistenceFactory(), RollingSelector.INSTANCE, applicationMessage);
-        }
-    }
-
-}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/metric/TraceSegmentRecordMember.java

@@ -1,7 +1,12 @@
-package com.a.eye.skywalking.collector.worker.persistence;
+package com.a.eye.skywalking.collector.worker.application.metric;
 
-import com.a.eye.skywalking.collector.actor.AbstractWorker;
-import com.a.eye.skywalking.collector.worker.RecordCollection;
+
+import akka.actor.ActorRef;
+import com.a.eye.skywalking.collector.actor.AbstractMember;
+import com.a.eye.skywalking.collector.actor.AbstractMemberProvider;
+import com.a.eye.skywalking.collector.actor.MemberSystem;
+import com.a.eye.skywalking.collector.actor.selector.LocalSelector;
+import com.a.eye.skywalking.collector.worker.application.persistence.TraceSegmentRecordPersistence;
 import com.a.eye.skywalking.trace.Span;
 import com.a.eye.skywalking.trace.TraceSegment;
 import com.a.eye.skywalking.trace.TraceSegmentRef;
@@ -14,17 +19,30 @@ import java.util.Map;
 /**
  * @author pengys5
  */
-public class AppTraceSegmentRecord extends AbstractWorker {
+public class TraceSegmentRecordMember extends AbstractMember {
 
-    private RecordCollection recordCollection = new RecordCollection();
+    public TraceSegmentRecordMember(MemberSystem memberSystem, ActorRef actorRef) {
+        super(memberSystem, actorRef);
+    }
+
+    @Override
+    public void preStart() throws Throwable {
+    }
 
     @Override
     public void receive(Object message) throws Throwable {
         if (message instanceof TraceSegment) {
             TraceSegment traceSegment = (TraceSegment) message;
+            JsonObject traceSegmentJsonObj = parseTraceSegment(traceSegment);
 
-            JsonObject traceJsonObj = parseTraceSegment(traceSegment);
-            recordCollection.put("", traceSegment.getTraceSegmentId(), traceJsonObj);
+            tell(new TraceSegmentRecordPersistence.Factory(), LocalSelector.INSTANCE, traceSegmentJsonObj);
+        }
+    }
+
+    public static class Factory extends AbstractMemberProvider {
+        @Override
+        public Class memberClass() {
+            return TraceSegmentRecordMember.class;
         }
     }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/ApplicationMessage.java

@@ -1,34 +0,0 @@
-package com.a.eye.skywalking.collector.worker.application.persistence;
-
-/**
- * @author pengys5
- */
-public class ApplicationMessage {
-    private final String code;
-    private final String component;
-    private final String host;
-    private final String layer;
-
-    public ApplicationMessage(String code, String component, String host, String layer) {
-        this.code = code;
-        this.component = component;
-        this.host = host;
-        this.layer = layer;
-    }
-
-    public String getCode() {
-        return code;
-    }
-
-    public String getComponent() {
-        return component;
-    }
-
-    public String getHost() {
-        return host;
-    }
-
-    public String getLayer() {
-        return layer;
-    }
-}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/ApplicationPersistence.java

@@ -1,25 +0,0 @@
-package com.a.eye.skywalking.collector.worker.application.persistence;
-
-import com.a.eye.skywalking.collector.worker.persistence.PersistenceMessage;
-import com.a.eye.skywalking.collector.worker.persistence.PersistenceWorker;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * @author pengys5
- */
-public class ApplicationPersistence extends PersistenceWorker<Object> {
-
-    private Map<String, ApplicationMessage> appData = new HashMap();
-
-    @Override
-    public void receive(Object message) throws Throwable {
-        if (message instanceof ApplicationMessage) {
-            ApplicationMessage applicationMessage = (ApplicationMessage) message;
-            appData.put(applicationMessage.getCode(), applicationMessage);
-        } else if (message instanceof PersistenceMessage) {
-            
-        }
-    }
-}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/ApplicationPersistenceFactory.java

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker.application.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-
-/**
- * @author pengys5
- */
-public class ApplicationPersistenceFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return ApplicationPersistence.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/DAGNodePersistence.java

@@ -0,0 +1,59 @@
+package com.a.eye.skywalking.collector.worker.application.persistence;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.PersistenceWorker;
+import com.google.gson.JsonObject;
+
+/**
+ * @author pengys5
+ */
+public class DAGNodePersistence extends PersistenceWorker<DAGNodePersistence.Metric> {
+
+    @Override
+    public String esIndex() {
+        return "application";
+    }
+
+    @Override
+    public String esType() {
+        return "dag_node";
+    }
+
+    @Override
+    public void analyse(Object message) throws Throwable {
+        if (message instanceof Metric) {
+            Metric metric = (Metric) message;
+            JsonObject propertyJsonObj = new JsonObject();
+            propertyJsonObj.addProperty("code", metric.code);
+            propertyJsonObj.addProperty("component", metric.component);
+            propertyJsonObj.addProperty("layer", metric.layer);
+
+            putData(metric.code, propertyJsonObj);
+        }
+    }
+
+    public static class Factory extends AbstractWorkerProvider {
+        @Override
+        public Class workerClass() {
+            return DAGNodePersistence.class;
+        }
+
+        @Override
+        public int workerNum() {
+            return WorkerConfig.WorkerNum.DAGNodePersistence_Num;
+        }
+    }
+
+    public static class Metric {
+        private final String code;
+        private final String component;
+        private final String layer;
+
+        public Metric(String code, String component, String layer) {
+            this.code = code;
+            this.component = component;
+            this.layer = layer;
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/NodeInstancePersistence.java

@@ -0,0 +1,56 @@
+package com.a.eye.skywalking.collector.worker.application.persistence;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.PersistenceWorker;
+import com.google.gson.JsonObject;
+
+/**
+ * @author pengys5
+ */
+public class NodeInstancePersistence extends PersistenceWorker<NodeInstancePersistence.Metric> {
+
+    @Override
+    public String esIndex() {
+        return "application";
+    }
+
+    @Override
+    public String esType() {
+        return "node_instance";
+    }
+
+    @Override
+    public void analyse(Object message) throws Throwable {
+        if (message instanceof Metric) {
+            Metric metric = (Metric) message;
+            JsonObject propertyJsonObj = new JsonObject();
+            propertyJsonObj.addProperty("code", metric.code);
+            propertyJsonObj.addProperty("address", metric.address);
+
+            putData(metric.address, propertyJsonObj);
+        }
+    }
+
+    public static class Factory extends AbstractWorkerProvider {
+        @Override
+        public Class workerClass() {
+            return NodeInstancePersistence.class;
+        }
+
+        @Override
+        public int workerNum() {
+            return WorkerConfig.WorkerNum.NodeInstancePersistence_Num;
+        }
+    }
+
+    public static class Metric {
+        private final String code;
+        private final String address;
+
+        public Metric(String code, String address) {
+            this.code = code;
+            this.address = address;
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/ResponseCostPersistence.java

@@ -0,0 +1,82 @@
+package com.a.eye.skywalking.collector.worker.application.persistence;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.PersistenceWorker;
+import com.google.gson.JsonObject;
+
+/**
+ * @author pengys5
+ */
+public class ResponseCostPersistence extends PersistenceWorker<ResponseCostPersistence.Metric> {
+
+    @Override
+    public String esIndex() {
+        return "application_metric";
+    }
+
+    @Override
+    public String esType() {
+        return "response_cost";
+    }
+
+    @Override
+    public void analyse(Object message) throws Throwable {
+        if (message instanceof Metric) {
+            Metric metric = (Metric) message;
+            long cost = metric.startTime - metric.endTime;
+            JsonObject data;
+            if (containsId(metric.code)) {
+                data = getData(metric.code);
+            } else {
+                data = new JsonObject();
+            }
+
+            String propertyKey = "";
+
+            if (cost <= 1000 && !metric.isError) {
+                propertyKey = "one_second_less";
+            } else if (cost > 1000 && cost <= 3000 && !metric.isError) {
+                propertyKey = "three_second_less";
+            } else if (cost > 3000 && cost <= 5000 && !metric.isError) {
+                propertyKey = "five_second_less";
+            } else if (cost > 5000 && cost <= 5000 && !metric.isError) {
+                propertyKey = "slow";
+            } else {
+                propertyKey = "error";
+            }
+
+            if (data.has(propertyKey)) {
+                data.addProperty(propertyKey, data.get(propertyKey).getAsLong() + 1);
+            } else {
+                data.addProperty(propertyKey, 1);
+            }
+        }
+    }
+
+    public static class Factory extends AbstractWorkerProvider {
+        @Override
+        public Class workerClass() {
+            return ResponseCostPersistence.class;
+        }
+
+        @Override
+        public int workerNum() {
+            return WorkerConfig.WorkerNum.ResponseCostPersistence_Num;
+        }
+    }
+
+    public static class Metric {
+        private final String code;
+        private final Boolean isError;
+        private final Long startTime;
+        private final Long endTime;
+
+        public Metric(String code, Boolean isError, Long startTime, Long endTime) {
+            this.code = code;
+            this.isError = isError;
+            this.startTime = startTime;
+            this.endTime = endTime;
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/ResponseSummaryPersistence.java

@@ -0,0 +1,72 @@
+package com.a.eye.skywalking.collector.worker.application.persistence;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
+import com.a.eye.skywalking.collector.worker.PersistenceWorker;
+import com.google.gson.JsonObject;
+
+import static com.a.eye.skywalking.collector.worker.WorkerConfig.WorkerNum.ResponseSummaryPersistence_Num;
+
+/**
+ * @author pengys5
+ */
+public class ResponseSummaryPersistence extends PersistenceWorker<ResponseSummaryPersistence.Metric> {
+
+    @Override
+    public String esIndex() {
+        return "application_metric";
+    }
+
+    @Override
+    public String esType() {
+        return "response_summary";
+    }
+
+    @Override
+    public void analyse(Object message) throws Throwable {
+        if (message instanceof Metric) {
+            Metric metric = (Metric) message;
+
+            JsonObject data;
+            if (containsId(metric.code)) {
+                data = getData(metric.code);
+            } else {
+                data = new JsonObject();
+            }
+
+            String propertyKey = "";
+            if (metric.isError) {
+                propertyKey = "error";
+            } else {
+                propertyKey = "success";
+            }
+
+            if (data.has(propertyKey)) {
+                data.addProperty(propertyKey, data.get(propertyKey).getAsLong() + 1);
+            } else {
+                data.addProperty(propertyKey, 1);
+            }
+        }
+    }
+
+    public static class Factory extends AbstractWorkerProvider {
+        @Override
+        public Class workerClass() {
+            return ResponseSummaryPersistence.class;
+        }
+
+        @Override
+        public int workerNum() {
+            return ResponseSummaryPersistence_Num;
+        }
+    }
+
+    public static class Metric {
+        private final String code;
+        private final Boolean isError;
+
+        public Metric(String code, Boolean isError) {
+            this.code = code;
+            this.isError = isError;
+        }
+    }
+}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/TraceSegmentRecordPersistence.java

@@ -0,0 +1,43 @@
+package com.a.eye.skywalking.collector.worker.application.persistence;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
+import com.a.eye.skywalking.collector.worker.PersistenceWorker;
+import com.google.gson.JsonObject;
+
+import static com.a.eye.skywalking.collector.worker.WorkerConfig.WorkerNum.TraceSegmentRecordPersistence_Num;
+
+/**
+ * @author pengys5
+ */
+public class TraceSegmentRecordPersistence extends PersistenceWorker {
+
+    @Override
+    public String esIndex() {
+        return "application_record";
+    }
+
+    @Override
+    public String esType() {
+        return "trace_segment";
+    }
+
+    @Override
+    public void analyse(Object message) throws Throwable {
+        if (message instanceof JsonObject) {
+            JsonObject traceSegmentJsonObj = (JsonObject) message;
+            putData(traceSegmentJsonObj.get("segmentId").getAsString(), traceSegmentJsonObj);
+        }
+    }
+
+    public static class Factory extends AbstractWorkerProvider {
+        @Override
+        public Class workerClass() {
+            return TraceSegmentRecordPersistence.class;
+        }
+
+        @Override
+        public int workerNum() {
+            return TraceSegmentRecordPersistence_Num;
+        }
+    }
+}

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

@@ -0,0 +1,44 @@
+package com.a.eye.skywalking.collector.worker.applicationref;
+
+import akka.actor.ActorRef;
+import com.a.eye.skywalking.collector.actor.AbstractMember;
+import com.a.eye.skywalking.collector.actor.AbstractMemberProvider;
+import com.a.eye.skywalking.collector.actor.MemberSystem;
+import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
+import com.a.eye.skywalking.collector.worker.applicationref.presistence.DAGNodeRefPersistence;
+import com.a.eye.skywalking.trace.TraceSegment;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationRefMember extends AbstractMember {
+
+    public ApplicationRefMember(MemberSystem memberSystem, ActorRef actorRef) {
+        super(memberSystem, actorRef);
+    }
+
+    @Override
+    public void preStart() throws Throwable {
+
+    }
+
+    @Override
+    public void receive(Object message) throws Throwable {
+        TraceSegment traceSegment = (TraceSegment) message;
+
+        if (traceSegment.getPrimaryRef() != null) {
+            String front = traceSegment.getPrimaryRef().getApplicationCode();
+            String behind = traceSegment.getApplicationCode();
+
+            DAGNodeRefPersistence.Metric nodeRef = new DAGNodeRefPersistence.Metric(front, behind);
+            tell(new DAGNodeRefPersistence.Factory(), RollingSelector.INSTANCE, nodeRef);
+        }
+    }
+
+    public static class Factory extends AbstractMemberProvider {
+        @Override
+        public Class memberClass() {
+            return ApplicationRefMember.class;
+        }
+    }
+}

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

@@ -1,13 +0,0 @@
-package com.a.eye.skywalking.collector.worker.applicationref;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorker;
-
-/**
- * @author pengys5
- */
-public class ApplicationRefWorker extends AbstractWorker {
-    @Override
-    public void receive(Object message) throws Throwable {
-
-    }
-}

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

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker.applicationref;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-
-/**
- * @author pengys5
- */
-public class ApplicationRefWorkerFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return ApplicationRefWorker.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/presistence/DAGNodeRefPersistence.java

@@ -0,0 +1,56 @@
+package com.a.eye.skywalking.collector.worker.applicationref.presistence;
+
+import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
+import com.a.eye.skywalking.collector.worker.PersistenceWorker;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.google.gson.JsonObject;
+
+/**
+ * @author pengys5
+ */
+public class DAGNodeRefPersistence extends PersistenceWorker<DAGNodeRefPersistence.Metric> {
+
+    @Override
+    public String esIndex() {
+        return "node_ref";
+    }
+
+    @Override
+    public String esType() {
+        return "node_ref";
+    }
+
+    @Override
+    public void analyse(Object message) throws Throwable {
+        if (message instanceof Metric) {
+            Metric metric = (Metric) message;
+            JsonObject propertyJsonObj = new JsonObject();
+            propertyJsonObj.addProperty("frontCode", metric.frontCode);
+            propertyJsonObj.addProperty("behindCode", metric.behindCode);
+
+            putData(metric.frontCode + "-" + metric.behindCode, propertyJsonObj);
+        }
+    }
+
+    public static class Factory extends AbstractWorkerProvider {
+        @Override
+        public Class workerClass() {
+            return DAGNodeRefPersistence.class;
+        }
+
+        @Override
+        public int workerNum() {
+            return WorkerConfig.WorkerNum.DAGNodeRefPersistence_Num;
+        }
+    }
+
+    public static class Metric {
+        private final String frontCode;
+        private final String behindCode;
+
+        public Metric(String frontCode, String behindCode) {
+            this.frontCode = frontCode;
+            this.behindCode = behindCode;
+        }
+    }
+}

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

@@ -1,39 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorker;
-import com.a.eye.skywalking.collector.worker.MetricCollection;
-
-/**
- * @author pengys5
- */
-public class AppResponseCost extends AbstractWorker {
-
-    private MetricCollection oneSecondsLessMetric = new MetricCollection();
-
-    private MetricCollection threeSecondsLessMetric = new MetricCollection();
-
-    private MetricCollection fiveSecondsLessMetric = new MetricCollection();
-
-    private MetricCollection slowSecondsLessMetric = new MetricCollection();
-
-    private MetricCollection errorSecondsLessMetric = new MetricCollection();
-
-    @Override
-    public void receive(Object message) throws Throwable {
-        if (message instanceof AppResponseSummaryMessage) {
-            AppResponseCostMessage costMessage = (AppResponseCostMessage) message;
-            long cost = costMessage.getEndTime() - costMessage.getStartTime();
-            if (cost <= 1000 && !costMessage.getError()) {
-                oneSecondsLessMetric.put(costMessage.getTimeSlice(), costMessage.getCode(), cost);
-            } else if (cost > 1000 && cost <= 3000 && !costMessage.getError()) {
-                threeSecondsLessMetric.put(costMessage.getTimeSlice(), costMessage.getCode(), cost);
-            } else if (cost > 3000 && cost <= 5000 && !costMessage.getError()) {
-                fiveSecondsLessMetric.put(costMessage.getTimeSlice(), costMessage.getCode(), cost);
-            } else if (cost > 5000 && cost <= 5000 && !costMessage.getError()) {
-                slowSecondsLessMetric.put(costMessage.getTimeSlice(), costMessage.getCode(), cost);
-            } else {
-                errorSecondsLessMetric.put(costMessage.getTimeSlice(), costMessage.getCode(), cost);
-            }
-        }
-    }
-}

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

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-
-/**
- * @author pengys5
- */
-public class AppResponseCostFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return AppResponseCost.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

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

@@ -1,37 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.worker.TimeSliceMessage;
-
-/**
- * @author pengys5
- */
-public class AppResponseCostMessage extends TimeSliceMessage {
-    private final String code;
-    private final Boolean isError;
-    private final Long startTime;
-    private final Long endTime;
-
-    public AppResponseCostMessage(String timeSlice, String code, Boolean isError, Long startTime, Long endTime) {
-        super(timeSlice);
-        this.code = code;
-        this.isError = isError;
-        this.startTime = startTime;
-        this.endTime = endTime;
-    }
-
-    public String getCode() {
-        return code;
-    }
-
-    public Boolean getError() {
-        return isError;
-    }
-
-    public Long getStartTime() {
-        return startTime;
-    }
-
-    public Long getEndTime() {
-        return endTime;
-    }
-}

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

@@ -1,29 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorker;
-import com.a.eye.skywalking.collector.worker.MetricCollection;
-
-/**
- * @author pengys5
- */
-public class AppResponseSummary extends AbstractWorker {
-
-    private MetricCollection summaryMetric = new MetricCollection();
-
-    private MetricCollection errorSummaryMetric = new MetricCollection();
-
-    private MetricCollection successSummaryMetric = new MetricCollection();
-
-    @Override
-    public void receive(Object message) throws Throwable {
-        if (message instanceof AppResponseSummaryMessage) {
-            AppResponseSummaryMessage summaryMessage = (AppResponseSummaryMessage) message;
-            summaryMetric.put(summaryMessage.getTimeSlice(), summaryMessage.getCode(), 1l);
-            if (summaryMessage.getError()) {
-                errorSummaryMetric.put(summaryMessage.getTimeSlice(), summaryMessage.getCode(), 1l);
-            } else {
-                successSummaryMetric.put(summaryMessage.getTimeSlice(), summaryMessage.getCode(), 1l);
-            }
-        }
-    }
-}

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

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-
-/**
- * @author pengys5
- */
-public class AppResponseSummaryFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return AppResponseSummary.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

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

@@ -1,25 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.worker.TimeSliceMessage;
-
-/**
- * @author pengys5
- */
-public class AppResponseSummaryMessage extends TimeSliceMessage {
-    private final String code;
-    private final Boolean isError;
-
-    public AppResponseSummaryMessage(String timeSlice, String code, Boolean isError) {
-        super(timeSlice);
-        this.code = code;
-        this.isError = isError;
-    }
-
-    public String getCode() {
-        return code;
-    }
-
-    public Boolean getError() {
-        return isError;
-    }
-}

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

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-
-/**
- * @author pengys5
- */
-public class AppTraceSegmentRecordFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return AppTraceSegmentRecord.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

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

@@ -1,12 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.worker.TimeSliceMessage;
-
-/**
- * @author pengys5
- */
-public class AppTraceSegmentRecordMessage extends TimeSliceMessage {
-    public AppTraceSegmentRecordMessage(String timeSlice) {
-        super(timeSlice);
-    }
-}

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

@@ -1,24 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorker;
-import com.a.eye.skywalking.collector.worker.RecordCollection;
-import com.a.eye.skywalking.collector.worker.application.persistence.ApplicationMessage;
-import com.google.gson.JsonObject;
-
-/**
- * @author pengys5
- */
-public class ApplicationRefRecord extends AbstractWorker {
-
-    private RecordCollection refRecord = new RecordCollection();
-
-    @Override
-    public void receive(Object message) throws Throwable {
-        if (message instanceof ApplicationMessage) {
-            ApplicationRefRecordMessage applicationMessage = (ApplicationRefRecordMessage) message;
-            refRecord.put("", applicationMessage.getCode() + "-" + applicationMessage.getRefCode(), new JsonObject());
-        } else if (message instanceof PersistenceMessage) {
-
-        }
-    }
-}

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

@@ -1,18 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-
-/**
- * @author pengys5
- */
-public class ApplicationRefRecordFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return ApplicationRefRecord.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

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

@@ -1,22 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-/**
- * @author pengys5
- */
-public class ApplicationRefRecordMessage {
-    private final String code;
-    private final String refCode;
-
-    public ApplicationRefRecordMessage(String code, String refCode) {
-        this.code = code;
-        this.refCode = refCode;
-    }
-
-    public String getCode() {
-        return code;
-    }
-
-    public String getRefCode() {
-        return refCode;
-    }
-}

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

@@ -1,7 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-/**
- * @author pengys5
- */
-public class PersistenceMessage {
-}

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

@@ -1,9 +0,0 @@
-package com.a.eye.skywalking.collector.worker.persistence;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorker;
-
-/**
- * @author pengys5
- */
-public abstract class PersistenceWorker<T> extends AbstractWorker<T> {
-}

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

@@ -1,6 +1,11 @@
 package com.a.eye.skywalking.collector.worker.receiver;
 
+import com.a.eye.skywalking.collector.actor.AbstractMember;
 import com.a.eye.skywalking.collector.actor.AbstractWorker;
+import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
+import com.a.eye.skywalking.collector.worker.application.ApplicationMember;
+import com.a.eye.skywalking.collector.worker.applicationref.ApplicationRefMember;
 import com.a.eye.skywalking.trace.TraceSegment;
 
 /**
@@ -8,10 +13,34 @@ import com.a.eye.skywalking.trace.TraceSegment;
  */
 public class TraceSegmentReceiver extends AbstractWorker {
 
+    @Override
+    public void preStart() throws Exception {
+        ApplicationMember.Factory factory = new ApplicationMember.Factory();
+        factory.createWorker(memberContext(), getSelf());
+    }
+
     @Override
     public void receive(Object message) throws Throwable {
         if (message instanceof TraceSegment) {
-            
+            TraceSegment traceSegment = (TraceSegment) message;
+
+            AbstractMember applicationMember = memberContext().memberFor(ApplicationMember.class.getSimpleName());
+            applicationMember.receive(traceSegment);
+
+            AbstractMember applicationRefMember = memberContext().memberFor(ApplicationRefMember.class.getSimpleName());
+            applicationRefMember.receive(traceSegment);
+        }
+    }
+
+    public class Factory extends AbstractWorkerProvider {
+        @Override
+        public Class workerClass() {
+            return TraceSegmentReceiver.class;
+        }
+
+        @Override
+        public int workerNum() {
+            return WorkerConfig.WorkerNum.TraceSegmentReceiver_Num;
         }
     }
 }

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

@@ -1,19 +0,0 @@
-package com.a.eye.skywalking.collector.worker.receiver;
-
-import com.a.eye.skywalking.collector.actor.AbstractWorkerProvider;
-import com.a.eye.skywalking.collector.worker.application.member.ApplicationDiscoverMember;
-
-/**
- * @author pengys5
- */
-public class TraceSegmentReceiverFactory extends AbstractWorkerProvider {
-    @Override
-    public Class workerClass() {
-        return ApplicationDiscoverMember.class;
-    }
-
-    @Override
-    public int workerNum() {
-        return 0;
-    }
-}

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

@@ -0,0 +1,30 @@
+package com.a.eye.skywalking.collector.worker.tools;
+
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.transport.client.PreBuiltTransportClient;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+/**
+ * @author pengys5
+ */
+public class EsClient {
+
+    private static TransportClient client;
+
+    public void boot() throws UnknownHostException {
+        Settings settings = Settings.builder()
+                .put("cluster.name", "myClusterName").build();
+
+        client = new PreBuiltTransportClient(Settings.EMPTY)
+                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("host1"), 9300))
+                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("host1"), 9300));
+    }
+
+    public static TransportClient client() {
+        return client;
+    }
+}