Merge branch 'feature/collector' of https://github.com/wu-sheng/sky-walking into feature/collector

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

@@ -41,8 +41,7 @@ public class CollectorSystem {
         final Config config = ConfigFactory.parseString("akka.remote.netty.tcp.hostname=" + ClusterConfig.Cluster.Current.hostname).
                 withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.port=" + ClusterConfig.Cluster.Current.port)).
                 withFallback(ConfigFactory.parseString("akka.cluster.roles=" + ClusterConfig.Cluster.Current.roles)).
-                withFallback(ConfigFactory.parseString("akka.actor.provider=" + ClusterConfig.Cluster.provider)).
-                withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + ClusterConfig.Cluster.nodes)).
+                withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + ClusterConfig.Cluster.seed_nodes)).
                 withFallback(ConfigFactory.load("application.conf"));
         ActorSystem akkaSystem = ActorSystem.create("ClusterSystem", config);
 

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

@@ -1,15 +1,12 @@
 package com.a.eye.skywalking.collector.cluster;
 
-import akka.actor.ActorSystem;
-
 /**
  * A static class contains some config values of cluster.
  * {@link Cluster.Current#hostname} is a ip address of server which start this process.
  * {@link Cluster.Current#port} is a port of server use to bind
  * {@link Cluster.Current#roles} is a roles of workers that use to create workers which
  * has those role in this process.
- * {@link Cluster#nodes} is a nodes which cluster have.
- * {@link Cluster#appname} is a name of {@link ActorSystem} in cluster.
+ * {@link Cluster#seed_nodes} is a seed_nodes which cluster have.
  *
  * @author pengys5
  */
@@ -22,9 +19,6 @@ public class ClusterConfig {
             public static String roles = "";
         }
 
-        public static String nodes = "127.0.0.1:2551";
-
-        public static final String appname = "CollectorSystem";
-        public static final String provider = "akka.cluster.ClusterActorRefProvider";
+        public static String seed_nodes = "127.0.0.1:2551";
     }
 }

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

@@ -53,8 +53,8 @@ public class ClusterConfigInitializer {
         if (!StringUtil.isEmpty(System.getProperty("cluster.current.roles"))) {
             ClusterConfig.Cluster.Current.roles = System.getProperty("cluster.current.roles");
         }
-        if (!StringUtil.isEmpty(System.getProperty("cluster.nodes"))) {
-            ClusterConfig.Cluster.nodes = System.getProperty("cluster.nodes");
+        if (!StringUtil.isEmpty(System.getProperty("cluster.seed_nodes"))) {
+            ClusterConfig.Cluster.seed_nodes = System.getProperty("cluster.seed_nodes");
         }
     }
 }

skywalking-collector/skywalking-collector-commons/src/main/java/com/a/eye/skywalking/collector/commons/serializer/TraceSegmentSerializer.java

@@ -1,6 +1,8 @@
 package com.a.eye.skywalking.collector.commons.serializer;
 
 import akka.serialization.JSerializer;
+import com.a.eye.skywalking.logging.ILog;
+import com.a.eye.skywalking.logging.LogManager;
 import com.a.eye.skywalking.trace.TraceSegment;
 import com.a.eye.skywalking.trace.proto.SegmentMessage;
 import com.google.protobuf.InvalidProtocolBufferException;
@@ -9,6 +11,8 @@ import com.google.protobuf.InvalidProtocolBufferException;
  * @author pengys5
  */
 public class TraceSegmentSerializer extends JSerializer {
+    private static ILog logger = LogManager.getLogger(TraceSegmentSerializer.class);
+
     @Override
     public boolean includeManifest() {
         return false;
@@ -31,7 +35,7 @@ public class TraceSegmentSerializer extends JSerializer {
         try {
             traceSegment = new TraceSegment(SegmentMessage.parseFrom(bytes));
         } catch (InvalidProtocolBufferException e) {
-            e.printStackTrace();
+            logger.warn("Can't covert message from byte[] to SegmentMessage");
         }
         return traceSegment;
     }

skywalking-collector/skywalking-collector-commons/src/main/resources/application.conf

@@ -5,19 +5,17 @@ akka {
     serializers {
       java = "akka.serialization.JavaSerializer"
       proto = "akka.remote.serialization.ProtobufSerializer"
-      data = "com.a.eye.skywalking.collector.worker.TraceSegmentSerializer"
+      TraceSegment = "com.a.eye.skywalking.collector.worker.TraceSegmentSerializer"
       json = "com.a.eye.skywalking.collector.commons.serializer.JsonSerializer"
     }
 
     serialization-bindings {
       "java.lang.String" = java
       "com.google.protobuf.Message" = proto
-      "com.a.eye.skywalking.messages.ISerializable" = data
+      "com.a.eye.skywalking.trace.TraceSegment" = TraceSegment
       "com.google.gson.JsonObject" = json
-      //      "java.io.Serializable" = none
     }
 
-    //    serialize-messages = on
     warn-about-java-serializer-usage = on
   }
 
@@ -25,8 +23,6 @@ akka {
     log-remote-lifecycle-events = off
 
     netty.tcp {
-      hostname = "127.0.0.1"
-      port = 1000
     }
   }
 
@@ -34,4 +30,4 @@ akka {
     auto-down-unreachable-after = off
     metrics.enabled = off
   }
-}
+}

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

@@ -26,8 +26,7 @@ public class CollectorBootStartUp {
         final Config config = ConfigFactory.parseString("akka.remote.netty.tcp.hostname=" + ClusterConfig.Cluster.Current.hostname).
                 withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.port=" + ClusterConfig.Cluster.Current.port)).
                 withFallback(ConfigFactory.parseString("akka.cluster.roles=" + ClusterConfig.Cluster.Current.roles)).
-                withFallback(ConfigFactory.parseString("akka.actor.provider=" + ClusterConfig.Cluster.provider)).
-                withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + ClusterConfig.Cluster.nodes)).
+                withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + ClusterConfig.Cluster.seed_nodes)).
                 withFallback(ConfigFactory.load("application.conf"));
 
 //        ActorSystem system = ActorSystem.create(ClusterConfig.Cluster.appname, config);

skywalking-collector/skywalking-collector-worker/src/main/resources/collector.config

@@ -1,4 +1,4 @@
 cluster.current.hostname=127.0.0.1
 cluster.current.port=1000
 cluster.current.roles=[WorkersListener, TraceSegmentReceiver, NodeInstancePersistence]
-cluster.nodes=["akka.tcp://CollectorSystem@127.0.0.1:1000", "akka.tcp://CollectorSystem@127.0.0.1:1001", "akka.tcp://CollectorSystem@127.0.0.1:1002"]
+cluster.seed_nodes=["akka.tcp://CollectorSystem@127.0.0.1:1000", "akka.tcp://CollectorSystem@127.0.0.1:1001", "akka.tcp://CollectorSystem@127.0.0.1:1002"]

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

@@ -28,11 +28,9 @@ public class StartUpTestCase {
         final Config config = ConfigFactory.parseString("akka.remote.netty.tcp.hostname=" + ClusterConfig.Cluster.Current.hostname).
                 withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.port=" + ClusterConfig.Cluster.Current.port)).
                 withFallback(ConfigFactory.parseString("akka.cluster.roles=" + ClusterConfig.Cluster.Current.roles)).
-                withFallback(ConfigFactory.parseString("akka.actor.provider=" + ClusterConfig.Cluster.provider)).
-                withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + ClusterConfig.Cluster.nodes)).
+                withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + ClusterConfig.Cluster.seed_nodes)).
                 withFallback(ConfigFactory.load("application.conf"));
-        ActorSystem system = ActorSystem.create(ClusterConfig.Cluster.appname, config);
-//        WorkersCreator.INSTANCE.boot(system);
+        ActorSystem system = ActorSystem.create("CollectorSystem", config);
 
         EsClient.boot();
 

skywalking-commons/skywalking-trace/src/main/java/com/a/eye/skywalking/messages/ISerializable.java

@@ -1,29 +0,0 @@
-package com.a.eye.skywalking.messages;
-
-import com.a.eye.skywalking.trace.TraceSegment;
-
-/**
- * All messages, which need to send between Akka actors, should implement this interface.
- * The whole {@link TraceSegment} supports this.
- *
- * T should be a protobuf Object, which is generated by protoc.
- * {@see /sky-walking/skywalking-commons/skywalking-trace/src/main/proto/trace.proto}
- *
- * {@see https://github.com/google/protobuf/tree/master/java}
- *
- * Created by wusheng on 2017/2/22.
- */
-public interface ISerializable<T> {
-    /**
-     * Serialize this object to T
-     * @return
-     */
-    T serialize();
-
-    /**
-     * Initialize this object by the given message.
-     *
-     * @param message to init object.
-     */
-    void deserialize(T message);
-}

skywalking-commons/skywalking-trace/src/main/java/com/a/eye/skywalking/trace/LogData.java

@@ -1,7 +1,6 @@
 package com.a.eye.skywalking.trace;
 
 import com.a.eye.skywalking.api.util.StringUtil;
-import com.a.eye.skywalking.messages.ISerializable;
 import com.a.eye.skywalking.trace.proto.KeyValue;
 import com.a.eye.skywalking.trace.proto.LogDataMessage;
 import java.util.Collections;
@@ -14,7 +13,7 @@ import java.util.Map;
  *
  * Created by wusheng on 2017/2/17.
  */
-public class LogData implements ISerializable<LogDataMessage> {
+public class LogData {
     private long time;
     private Map<String, ?> fields;
 
@@ -38,7 +37,6 @@ public class LogData implements ISerializable<LogDataMessage> {
         return Collections.unmodifiableMap(fields);
     }
 
-    @Override
     public LogDataMessage serialize() {
         LogDataMessage.Builder logDataBuilder = LogDataMessage.newBuilder();
         logDataBuilder.setTime(time);
@@ -59,7 +57,6 @@ public class LogData implements ISerializable<LogDataMessage> {
         return logDataBuilder.build();
     }
 
-    @Override
     public void deserialize(LogDataMessage message) {
         time = message.getTime();
         List<KeyValue> list = message.getFieldsList();

skywalking-commons/skywalking-trace/src/main/java/com/a/eye/skywalking/trace/Span.java

@@ -1,7 +1,6 @@
 package com.a.eye.skywalking.trace;
 
 import com.a.eye.skywalking.api.util.StringUtil;
-import com.a.eye.skywalking.messages.ISerializable;
 import com.a.eye.skywalking.trace.proto.KeyValue;
 import com.a.eye.skywalking.trace.proto.LogDataMessage;
 import com.a.eye.skywalking.trace.proto.SpanMessage;
@@ -23,7 +22,7 @@ import java.util.Map;
  *
  * Created by wusheng on 2017/2/17.
  */
-public class Span implements ISerializable<SpanMessage> {
+public class Span{
     private int spanId;
 
     private int parentSpanId;
@@ -266,7 +265,6 @@ public class Span implements ISerializable<SpanMessage> {
         return log(exceptionFields);
     }
 
-    @Override
     public SpanMessage serialize() {
         SpanMessage.Builder builder = SpanMessage.newBuilder();
         builder.setSpanId(spanId);
@@ -290,7 +288,6 @@ public class Span implements ISerializable<SpanMessage> {
         return builder.build();
     }
 
-    @Override
     public void deserialize(SpanMessage message) {
         spanId = message.getSpanId();
         parentSpanId = message.getParentSpanId();

skywalking-commons/skywalking-trace/src/main/java/com/a/eye/skywalking/trace/TraceId/DistributedTraceId.java

@@ -23,4 +23,21 @@ public abstract class DistributedTraceId {
     public String get() {
         return id;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+        if (o == null || getClass() != o.getClass())
+            return false;
+
+        DistributedTraceId id1 = (DistributedTraceId)o;
+
+        return id != null ? id.equals(id1.id) : id1.id == null;
+    }
+
+    @Override
+    public int hashCode() {
+        return id != null ? id.hashCode() : 0;
+    }
 }

skywalking-commons/skywalking-trace/src/main/java/com/a/eye/skywalking/trace/TraceSegment.java

@@ -1,6 +1,5 @@
 package com.a.eye.skywalking.trace;
 
-import com.a.eye.skywalking.messages.ISerializable;
 import com.a.eye.skywalking.trace.TraceId.DistributedTraceId;
 import com.a.eye.skywalking.trace.TraceId.NewDistributedTraceId;
 import com.a.eye.skywalking.trace.TraceId.PropagatedTraceId;
@@ -21,7 +20,7 @@ import java.util.List;
  *
  * Created by wusheng on 2017/2/17.
  */
-public class TraceSegment implements ISerializable<SegmentMessage> {
+public class TraceSegment {
     private static final String ID_TYPE = "Segment";
 
     /**
@@ -191,7 +190,6 @@ public class TraceSegment implements ISerializable<SegmentMessage> {
             '}';
     }
 
-    @Override
     public SegmentMessage serialize() {
         SegmentMessage.Builder segmentBuilder = SegmentMessage.newBuilder();
         segmentBuilder.setTraceSegmentId(traceSegmentId);
@@ -212,7 +210,6 @@ public class TraceSegment implements ISerializable<SegmentMessage> {
         return segmentBuilder.build();
     }
 
-    @Override
     public void deserialize(SegmentMessage message) {
         traceSegmentId = message.getTraceSegmentId();
         startTime = message.getStartTime();

skywalking-commons/skywalking-trace/src/main/java/com/a/eye/skywalking/trace/TraceSegmentRef.java

@@ -1,6 +1,5 @@
 package com.a.eye.skywalking.trace;
 
-import com.a.eye.skywalking.messages.ISerializable;
 import com.a.eye.skywalking.trace.proto.SegmentRefMessage;
 import com.a.eye.skywalking.trace.tag.Tags;
 
@@ -10,7 +9,7 @@ import com.a.eye.skywalking.trace.tag.Tags;
  *
  * Created by wusheng on 2017/2/17.
  */
-public class TraceSegmentRef implements ISerializable<SegmentRefMessage> {
+public class TraceSegmentRef{
     /**
      * {@link TraceSegment#traceSegmentId}
      */
@@ -78,7 +77,6 @@ public class TraceSegmentRef implements ISerializable<SegmentRefMessage> {
             '}';
     }
 
-    @Override
     public SegmentRefMessage serialize() {
         SegmentRefMessage.Builder builder = SegmentRefMessage.newBuilder();
         builder.setTraceSegmentId(traceSegmentId);
@@ -90,7 +88,6 @@ public class TraceSegmentRef implements ISerializable<SegmentRefMessage> {
         return builder.build();
     }
 
-    @Override
     public void deserialize(SegmentRefMessage message) {
         traceSegmentId = message.getTraceSegmentId();
         spanId = message.getSpanId();
@@ -107,21 +104,15 @@ public class TraceSegmentRef implements ISerializable<SegmentRefMessage> {
 
         TraceSegmentRef ref = (TraceSegmentRef)o;
 
-        if (spanId != ref.spanId)
-            return false;
         if (traceSegmentId != null ? !traceSegmentId.equals(ref.traceSegmentId) : ref.traceSegmentId != null)
             return false;
-        if (applicationCode != null ? !applicationCode.equals(ref.applicationCode) : ref.applicationCode != null)
-            return false;
-        return peerHost != null ? peerHost.equals(ref.peerHost) : ref.peerHost == null;
+        return applicationCode != null ? applicationCode.equals(ref.applicationCode) : ref.applicationCode == null;
     }
 
     @Override
     public int hashCode() {
         int result = traceSegmentId != null ? traceSegmentId.hashCode() : 0;
-        result = 31 * result + spanId;
         result = 31 * result + (applicationCode != null ? applicationCode.hashCode() : 0);
-        result = 31 * result + (peerHost != null ? peerHost.hashCode() : 0);
         return result;
     }
 }

skywalking-sniffer/skywalking-api/pom.xml

@@ -30,11 +30,6 @@
             <artifactId>skywalking-logging-api</artifactId>
             <version>${project.version}</version>
         </dependency>
-        <dependency>
-            <groupId>com.a.eye</groupId>
-            <artifactId>skywalking-collector-cluster</artifactId>
-            <version>${project.version}</version>
-        </dependency>
         <dependency>
             <groupId>com.a.eye</groupId>
             <artifactId>skywalking-collector-commons</artifactId>