增加路由信息

skywalking-network/src/main/proto/Spans.proto

@@ -14,7 +14,7 @@ message AckSpan {
     string userId = 8;
     string applicationId = 9;
 
-    int32 routeKey = 10;
+    int64 routeKey = 10;
 }
 
 message RequestSpan {
@@ -32,7 +32,7 @@ message RequestSpan {
     int32 processNo = 13;
     string address = 14;
 
-    int32 routeKey = 15;
+    int64 routeKey = 15;
 }
 
 message TraceId{

skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/invoke/monitor/RPCClientInvokeMonitor.java

@@ -27,7 +27,7 @@ public class RPCClientInvokeMonitor extends BaseInvokeMonitor {
 
             sendRequestSpan(spanData, id);
 
-            return new ContextData(spanData.getTraceId(), generateSubParentLevelId(spanData));
+            return new ContextData(spanData.getTraceId(), generateSubParentLevelId(spanData), spanData.getRouteKey());
         } catch (Throwable t) {
             logger.error(t.getMessage(), t);
             return new EmptyContextData();

skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/model/ContextData.java

@@ -8,26 +8,29 @@ public class ContextData {
     private TraceId traceId;
     private String parentLevel;
     private int levelId;
+    private long routeKey;
 
     ContextData() {
 
     }
 
-    public ContextData(TraceId traceId, String parentLevel) {
+    public ContextData(TraceId traceId, String parentLevelId, long routeKey) {
         this.traceId = traceId;
-        this.parentLevel = parentLevel;
+        this.parentLevel = parentLevelId;
+        this.routeKey = routeKey;
     }
 
     public ContextData(Span span) {
         this.traceId = span.getTraceId();
         this.parentLevel = span.getParentLevel();
         this.levelId = span.getLevelId();
+        this.routeKey = span.getRouteKey();
     }
 
     public ContextData(String contextDataStr) {
         // 反序列化参数
         String[] value = contextDataStr.split("-");
-        if (value == null || value.length != 3) {
+        if (value == null || value.length != 4) {
             throw new IllegalArgumentException("illegal context");
         }
         String traceIdStr = value[0];
@@ -48,6 +51,7 @@ public class ContextData {
         this.traceId = traceIdBuilder.build();
         this.parentLevel = value[1].trim();
         this.levelId = Integer.valueOf(value[2]);
+        this.routeKey = Long.parseLong(value[3]);
     }
 
     public TraceId getTraceId() {
@@ -75,6 +79,8 @@ public class ContextData {
         }
         stringBuilder.append("-");
         stringBuilder.append(levelId);
+        stringBuilder.append("-");
+        stringBuilder.append(routeKey);
         return stringBuilder.toString();
     }
 }

skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/model/Span.java

@@ -67,6 +67,7 @@ public class Span {
      */
     private String userId;
     private String viewPointId;
+    private long routeKey;
 
     public Span(TraceId traceId, String applicationId, String userId) {
         this.traceId = traceId;
@@ -197,7 +198,7 @@ public class Span {
 
     public RequestSpan.Builder buildRequestSpan(RequestSpan.Builder builder) {
         builder.setTraceId(this.traceId).setParentLevel(this.parentLevel).setLevelId(this.levelId).setSpanType(this.spanType).setApplicationId(this.applicationId)
-                .setUserId(this.userId);
+                .setUserId(this.userId).setRouteKey(routeKey);
         return builder;
     }
 
@@ -205,8 +206,15 @@ public class Span {
         builder.setTraceId(this.traceId).setParentLevel(this.parentLevel).setLevelId(this.levelId)
                 .setCost(System.currentTimeMillis() - this.startDate).setStatusCode(this.statusCode)
                 .setExceptionStack(this.exceptionStack).setUserId(this.userId).setApplicationId(this.applicationId)
-                .setViewpointId(this.viewPointId);
+                .setViewpointId(this.viewPointId).setRouteKey(routeKey);
         return builder;
     }
 
+    public void setRouteKey(long routeKey) {
+        this.routeKey = routeKey;
+    }
+
+    public long getRouteKey() {
+        return routeKey;
+    }
 }

skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/util/ContextGenerator.java

@@ -14,9 +14,7 @@ public final class ContextGenerator {
      * @return
      */
     public static Span generateSpanFromThreadLocal(Identification id) {
-        Span spanData = getSpanFromThreadLocal();
-        spanData.setStartDate(System.currentTimeMillis());
-        spanData.setViewPointId(id.getViewPoint());
+        Span spanData = getSpanFromThreadLocal(id);
         return spanData;
     }
 
@@ -32,21 +30,22 @@ public final class ContextGenerator {
         if (context != null && context.getTraceId() != null && spanData == null){
             spanData = new Span(context.getTraceId(), context.getParentLevel(), context.getLevelId(), Config.SkyWalking.APPLICATION_CODE, Config.SkyWalking.USER_ID);
         }else{
-            spanData = getSpanFromThreadLocal();
+            spanData = getSpanFromThreadLocal(id);
         }
-        spanData.setStartDate(System.currentTimeMillis());
-        spanData.setViewPointId(id.getViewPoint());
+
         return spanData;
     }
 
-    private static Span getSpanFromThreadLocal() {
+    private static Span getSpanFromThreadLocal(Identification id) {
         Span span;
         // 1.获取Context，从ThreadLocal栈中获取中
         final Span parentSpan = CurrentThreadSpanStack.peek();
         // 2 校验Context，Context是否存在
+        long routeKey = 0;
         if (parentSpan == null) {
             // 不存在，新创建一个Context
             span = new Span(TraceIdGenerator.generate(), Config.SkyWalking.APPLICATION_CODE, Config.SkyWalking.USER_ID);
+            routeKey = TokenGenerator.generate(id.getViewPoint());
         } else {
 
             // 根据ParentContextData的TraceId和RPCID
@@ -58,7 +57,13 @@ public final class ContextGenerator {
             } else {
                 span.setParentLevel(String.valueOf(parentSpan.getLevelId()));
             }
+            routeKey = parentSpan.getRouteKey();
         }
+
+        span.setStartDate(System.currentTimeMillis());
+        span.setViewPointId(id.getViewPoint());
+        span.setRouteKey(routeKey);
+
         return span;
     }
 

skywalking-sniffer/skywalking-api/src/main/java/com/a/eye/skywalking/util/TokenGenerator.java

+package com.a.eye.skywalking.util;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
+/**
+ * Created data xin on 2016/12/4.
+ */
+public class TokenGenerator {
+
+    public static long generate(String originData) {
+        long value = 0;
+        try {
+            MessageDigest messageDigest = MessageDigest.getInstance("MD5");
+            messageDigest.update(originData.getBytes());
+            byte[] data = messageDigest.digest();
+            //
+            for (int i = 0; i < data.length; i++) {
+                value = (value << 8) + (data[i] & 0xff);
+            }
+
+        } catch (NoSuchAlgorithmException e) {
+            e.printStackTrace();
+        }
+
+        return value;
+    }
+}

skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/router/Router.java

@@ -10,7 +10,10 @@ import com.a.eye.skywalking.routing.disruptor.NoopSpanDisruptor;
 import com.a.eye.skywalking.routing.disruptor.SpanDisruptor;
 import com.a.eye.skywalking.routing.storage.listener.NodeChangesListener;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 
 public class Router implements NodeChangesListener {
     private static ILog logger = LogManager.getLogger(Router.class);
@@ -26,13 +29,13 @@ public class Router implements NodeChangesListener {
         return getSpanDisruptor(ackSpan.getRouteKey());
     }
 
-    private SpanDisruptor getSpanDisruptor(int routKey) {
+    private SpanDisruptor getSpanDisruptor(long routKey) {
         if (disruptors.length == 0) {
             return noopSpanPool;
         }
 
         while (true) {
-            int index = routKey % disruptors.length;
+            int index = Math.abs((int) (routKey % disruptors.length));
             try {
                 return disruptors[index];
             } catch (ArrayIndexOutOfBoundsException e) {