feat:support traffic mirroring. (#623)

Author: SkyeBeFreeman

polaris-common/polaris-config-default/src/main/resources/conf/default-config.yml

@@ -197,6 +197,8 @@ consumer:
       - laneRouter
     #描述: 服务路由链
     chain:
+      # 流量镜像路由
+      - trafficMirroringRouter
       # 命名空间就近路由
       - namespaceRouter
       # 元数据路由

polaris-common/polaris-config/src/main/java/com/tencent/polaris/api/config/consumer/ServiceRouterConfig.java

@@ -48,6 +48,7 @@ public interface ServiceRouterConfig extends PluginConfig, Verifier {
 
     String DEFAULT_ROUTER_NAMESPACE = "namespaceRouter";
 
+    String DEFAULT_ROUTER_TRAFFIC_MIRRORING = "trafficMirroringRouter";
 
     /**
      * services.consumer.serviceRouter.beforeChain

polaris-common/polaris-model/src/main/java/com/tencent/polaris/api/pojo/ServiceEventKey.java

@@ -52,6 +52,7 @@ public enum EventType {
         NEARBY_ROUTE_RULE,
         LOSSLESS,
         BLOCK_ALLOW_RULE,
+        TRAFFIC_MIRRORING,
     }
 
     private final ServiceKey serviceKey;

polaris-common/polaris-model/src/main/java/com/tencent/polaris/api/utils/StringUtils.java

@@ -564,4 +564,11 @@ static int indexOf(final CharSequence cs, final CharSequence searchChar, final i
         return cs.toString().indexOf(searchChar.toString(), start);
     }
 
+
+    public static boolean contains(String str, String searchStr) {
+        if (str == null || searchStr == null) {
+            return false;
+        }
+        return str.indexOf(searchStr) >= 0;
+    }
 }

polaris-dependencies/pom.xml

@@ -434,6 +434,11 @@
                 <artifactId>router-canary</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.tencent.polaris</groupId>
+                <artifactId>router-mirroring</artifactId>
+                <version>${project.version}</version>
+            </dependency>
 
             <!-- polaris-plugins-location-->
             <dependency>

polaris-discovery/polaris-discovery-client/src/main/java/com/tencent/polaris/discovery/client/stat/InstancesStatisticUpdater.java

@@ -17,23 +17,20 @@
 
 package com.tencent.polaris.discovery.client.stat;
 
-import static com.tencent.polaris.api.pojo.RetStatus.RetSuccess;
-
 import com.tencent.polaris.api.plugin.registry.LocalRegistry;
 import com.tencent.polaris.api.plugin.registry.ResourceFilter;
-import com.tencent.polaris.api.pojo.Instance;
-import com.tencent.polaris.api.pojo.InstanceGauge;
-import com.tencent.polaris.api.pojo.InstanceStatistic;
-import com.tencent.polaris.api.pojo.ServiceEventKey;
+import com.tencent.polaris.api.pojo.*;
 import com.tencent.polaris.api.pojo.ServiceEventKey.EventType;
-import com.tencent.polaris.api.pojo.ServiceInstances;
-import com.tencent.polaris.api.pojo.ServiceKey;
 import com.tencent.polaris.api.utils.CollectionUtils;
+import com.tencent.polaris.api.utils.StringUtils;
 import com.tencent.polaris.client.pojo.InstanceByProto;
 import com.tencent.polaris.logging.LoggerFactory;
-import java.util.List;
 import org.slf4j.Logger;
 
+import java.util.List;
+
+import static com.tencent.polaris.api.pojo.RetStatus.RetSuccess;
+
 /**
  * InstancesStatisticUpdater
  *
@@ -53,12 +50,16 @@ public InstancesStatisticUpdater(LocalRegistry localRegistry) {
 
     public void updateInstanceStatistic(InstanceGauge result) {
         ServiceKey serviceKey = new ServiceKey(result.getNamespace(), result.getService());
-        ServiceEventKey serviceEventKey = new ServiceEventKey(serviceKey, EventType.INSTANCE);
-        ServiceInstances serviceInstances = localRegistry.getInstances(new ResourceFilter(serviceEventKey, true, true));
+        // 如果服务名或命名空间为空，则不统计
+        if (StringUtils.isBlank(serviceKey.getNamespace()) || StringUtils.isBlank(serviceKey.getService())) {
+            return;
+        }
         // 如果调用的是北极星内部的服务，则不统计
         if (serviceKey.getNamespace().equals(POLARIS_NAMESPACE)) {
             return;
         }
+        ServiceEventKey serviceEventKey = new ServiceEventKey(serviceKey, EventType.INSTANCE);
+        ServiceInstances serviceInstances = localRegistry.getInstances(new ResourceFilter(serviceEventKey, true, true));
         if (serviceInstances == null) {
             LOG.warn("[InstanceStatisticUpdater]: " + "service: " + serviceKey.getService() + " in namespace: "
                     + serviceKey.getNamespace() + " not found");

polaris-plugins/polaris-plugin-api/src/main/java/com/tencent/polaris/api/plugin/route/RouterConstants.java

@@ -29,4 +29,6 @@ public interface RouterConstants {
 
     // 服务路由容错
     String ROUTER_FAULT_TOLERANCE_ENABLE = "internal-enable-router-faulttolerance";
+
+    String TRAFFIC_MIRRORING_NODE_KEY = "internal-traffic-mirroring-node";
 }

polaris-plugins/polaris-plugins-connector/connector-composite/src/main/java/com/tencent/polaris/plugins/connector/composite/CompositeServiceUpdateTask.java

@@ -131,7 +131,8 @@ public void execute() {
                 || serviceEventKey.getEventType().equals(EventType.CIRCUIT_BREAKING)
                 || serviceEventKey.getEventType().equals(EventType.RATE_LIMITING)
                 || serviceEventKey.getEventType().equals(EventType.LANE_RULE)
-                || serviceEventKey.getEventType().equals(EventType.BLOCK_ALLOW_RULE)))) {
+                || serviceEventKey.getEventType().equals(EventType.BLOCK_ALLOW_RULE)
+                || serviceEventKey.getEventType().equals(EventType.TRAFFIC_MIRRORING)))) {
             return;
         }
 

polaris-plugins/polaris-plugins-connector/connector-consul/src/main/java/com/tencent/polaris/plugins/connector/consul/ConsulAPIConnector.java

@@ -56,6 +56,7 @@
 import com.tencent.polaris.plugins.connector.consul.service.lane.LaneService;
 import com.tencent.polaris.plugins.connector.consul.service.lossless.LosslessService;
 import com.tencent.polaris.plugins.connector.consul.service.ratelimiting.RateLimitingService;
+import com.tencent.polaris.plugins.connector.consul.service.router.MirroringService;
 import com.tencent.polaris.plugins.connector.consul.service.router.NearByRouteRuleService;
 import com.tencent.polaris.plugins.connector.consul.service.router.RoutingService;
 import org.slf4j.Logger;
@@ -255,6 +256,7 @@ private void initActually(InitContext ctx, ServerConnectorConfig connectorConfig
         consulServiceMap.put(ServiceEventKey.EventType.RATE_LIMITING, new RateLimitingService(consulClient, consulRawClient, consulContext, "consul-rate-limiting", mapper));
         consulServiceMap.put(ServiceEventKey.EventType.LANE_RULE, new LaneService(consulClient, consulRawClient, consulContext, "consul-lane", mapper));
         consulServiceMap.put(ServiceEventKey.EventType.BLOCK_ALLOW_RULE, new AuthorityService(consulClient, consulRawClient, consulContext, "consul-auth", mapper));
+        consulServiceMap.put(ServiceEventKey.EventType.TRAFFIC_MIRRORING, new MirroringService(consulClient, consulRawClient, consulContext, "consul-mirror", mapper));
         initialized = true;
     }