0.4.9-pre (#325)

* Feat/sync support2.x#mutiple thread sync02 (#304) * update port * Multithreading sync * solve conflict * imple SyncService * adapter deregister * optimization some code * fix deregister instance equals logic Co-authored-by: Oliver <wqdyxnbd@163.com> Co-authored-by: paderlol <huangmnlove@163.com> * Optimize the code for assigning tasks. (#320) * Develop (#321) * Optimize the code for assigning tasks. * Adds prefix to the input string if it doesn't already have it.#308 * Fix #305 (#322) * Optimize the code for assigning tasks. * Adds prefix to the input string if it doesn't already have it.#308 * Fix .#305 * Fix cyclic dependency code (#323) * Optimize the code for assigning tasks. * Adds prefix to the input string if it doesn't already have it.#308 * Fix .#305 * Fix cyclic dependency code. * Refactoring the Nacos Sync to Consul Logic (#324) * Optimize the code for assigning tasks. * Adds prefix to the input string if it doesn't already have it.#308 * Fix .#305 * Fix cyclic dependency code. * Refactoring the Nacos Sync to Consul Logic. --------- Co-authored-by: chenhao26 <35129699+chenhao26-nineteen@users.noreply.github.com> Co-authored-by: Oliver <wqdyxnbd@163.com>
nacos-group · May 15, 2023 · a1d683a · a1d683a
1 parent 61d3476
commit a1d683a
Show file tree

Hide file tree

Showing 36 changed files with 1,337 additions and 262 deletions.
diff --git a/nacossync-distribution/bin/nacosSync.sql b/nacossync-distribution/bin/nacosSync.sql
@@ -10,6 +10,8 @@ CREATE TABLE `cluster` (
   `connect_key_list` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
   `user_name` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
   `password` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
+  `namespace` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
+  `cluster_level` int default 0,
   PRIMARY KEY (`id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
 /******************************************/
@@ -39,5 +41,6 @@ CREATE TABLE `task` (
   `task_status` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
   `version` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
   `worker_ip` varchar(255) COLLATE utf8mb4_bin DEFAULT NULL,
+  `status` int default null ,
   PRIMARY KEY (`id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
diff --git a/nacossync-worker/pom.xml b/nacossync-worker/pom.xml
@@ -71,6 +71,10 @@
             <groupId>io.springfox</groupId>
             <artifactId>springfox-swagger-ui</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+        </dependency>
         <!--nacos-->
         <dependency>
             <groupId>com.alibaba.nacos</groupId>
@@ -161,6 +165,14 @@
                     </archive>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>9</source>
+                    <target>9</target>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/cache/SkyWalkerCacheServices.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/cache/SkyWalkerCacheServices.java
@@ -96,10 +96,25 @@ public FinishedTask getFinishedTask(TaskDO taskDO) {
 
         return finishedTaskMap.get(operationId);
     }
+
+    public FinishedTask getFinishedTask(String operationId) {
+        if (StringUtils.isEmpty(operationId)) {
+            return null;
+        }
+        return finishedTaskMap.get(operationId);
+    }
+
+    public FinishedTask removeFinishedTask(String operationId) {
+        if (StringUtils.isEmpty(operationId)) {
+            return null;
+        }
+        return finishedTaskMap.remove(operationId);
+    }
 
     public Map<String, FinishedTask> getFinishedTaskMap() {
 
         return finishedTaskMap;
     }
+
 
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/constant/SkyWalkerConstants.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/constant/SkyWalkerConstants.java
@@ -35,5 +35,7 @@ public class SkyWalkerConstants {
     public static  final String GROUP_NAME_PARAM="groupNameParam";
     public static  final String PAGE_NO="pageNo";
     public static  final String PAGE_SIZE="pageSize";
+    public static final String SYNC_INSTANCE_TAG="sync.instance.tag";
+    public static final Integer MAX_THREAD_NUM = 200;
 
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/dao/ClusterAccessService.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/dao/ClusterAccessService.java
@@ -103,4 +103,12 @@ private List<Predicate> getPredicates(Root<ClusterDO> root, CriteriaBuilder crit
         predicates.add(criteriaBuilder.like(root.get("clusterName"), "%" + queryCondition.getServiceName() + "%"));
         return predicates;
     }
+
+    public int findClusterLevel(String sourceClusterId){
+        ClusterDO clusterDO = clusterRepository.findByClusterId(sourceClusterId);
+        if (clusterDO != null) {
+            return clusterDO.getClusterLevel();
+        }
+        return -1;
+    }
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/dao/TaskAccessService.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/dao/TaskAccessService.java
@@ -114,5 +114,9 @@ private Page<TaskDO> getTaskDOS(QueryCondition queryCondition, Pageable pageable
 
                 }, pageable);
     }
+
+    public List<TaskDO> findServiceNameIsNull() {
+        return taskRepository.findAllByServiceNameEquals("ALL");
+    }
 
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/dao/repository/TaskRepository.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/dao/repository/TaskRepository.java
@@ -41,5 +41,12 @@ public interface TaskRepository extends CrudRepository<TaskDO, Integer>, JpaRepo
     List<TaskDO> findAllByTaskIdIn(List<String> taskIds);
 
     List<TaskDO> getAllByWorkerIp(String workerIp);
+
+    /**
+     * query service is all，use ns leven sync data
+     * @param serviceName
+     * @return
+     */
+    List<TaskDO> findAllByServiceNameEquals(String serviceName);
 
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/event/listener/EventListener.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/event/listener/EventListener.java
@@ -62,7 +62,7 @@ public void listenerSyncTaskEvent(SyncTaskEvent syncTaskEvent) {
 
         try {
             long start = System.currentTimeMillis();
-            if (syncManagerService.sync(syncTaskEvent.getTaskDO())) {                
+            if (syncManagerService.sync(syncTaskEvent.getTaskDO(), null)) {
                 skyWalkerCacheServices.addFinishedTask(syncTaskEvent.getTaskDO());
                 metricsManager.record(MetricsStatisticsType.SYNC_TASK_RT, System.currentTimeMillis() - start);
             } else {
@@ -88,7 +88,5 @@ public void listenerDeleteTaskEvent(DeleteTaskEvent deleteTaskEvent) {
         } catch (Exception e) {
             log.warn("listenerDeleteTaskEvent process error", e);
         }
-
     }
-
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/SyncManagerService.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/SyncManagerService.java
@@ -52,9 +52,9 @@ public boolean delete(TaskDO taskDO) throws NacosException {
 
     }
 
-    public boolean sync(TaskDO taskDO) {
+    public boolean sync(TaskDO taskDO, Integer index) {
 
-        return getSyncService(taskDO.getSourceClusterId(), taskDO.getDestClusterId()).sync(taskDO);
+        return getSyncService(taskDO.getSourceClusterId(), taskDO.getDestClusterId()).sync(taskDO, index);
 
     }
 

diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/SyncService.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/SyncService.java
@@ -35,15 +35,18 @@ public interface SyncService {
      * execute sync
      *
      * @param taskDO
+     * @param index
      * @return
      */
-    boolean sync(TaskDO taskDO);
+    boolean sync(TaskDO taskDO, Integer index);
 
     /**
      * Determines that the current instance data is from another source cluster
      */
     default boolean needSync(Map<String, String> sourceMetaData) {
-        return StringUtils.isBlank(sourceMetaData.get(SkyWalkerConstants.SOURCE_CLUSTERID_KEY));
+        boolean syncTag = StringUtils.isBlank(sourceMetaData.get(SkyWalkerConstants.SYNC_INSTANCE_TAG));
+        boolean blank = StringUtils.isBlank(sourceMetaData.get(SkyWalkerConstants.SOURCE_CLUSTERID_KEY));
+        return syncTag && blank;
     }
 
     /**

diff --git a/...worker/src/main/java/com/alibaba/nacossync/extension/holder/AbstractServerHolderImpl.java b/...worker/src/main/java/com/alibaba/nacossync/extension/holder/AbstractServerHolderImpl.java
@@ -26,7 +26,8 @@
 @Slf4j
 public abstract class AbstractServerHolderImpl<T> implements Holder {
 
-    private final Map<String, T> serviceMap = new ConcurrentHashMap<>();
+    protected final Map<String, T> serviceMap = new ConcurrentHashMap<>();
+
     @Autowired
     protected SkyWalkerCacheServices skyWalkerCacheServices;
 

diff --git a/...ssync-worker/src/main/java/com/alibaba/nacossync/extension/holder/EurekaServerHolder.java b/...ssync-worker/src/main/java/com/alibaba/nacossync/extension/holder/EurekaServerHolder.java
@@ -10,6 +10,7 @@
  * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
  * specific language governing permissions and limitations under the License.
  */
+
 package com.alibaba.nacossync.extension.holder;
 
 import com.alibaba.nacossync.extension.eureka.EurekaNamingService;
@@ -30,12 +31,27 @@
 @Service
 @Slf4j
 public class EurekaServerHolder extends AbstractServerHolderImpl<EurekaNamingService> {
+
+    private static final String HTTP_PREFIX = "http://";
+
+    private static final String HTTPS_PREFIX = "https://";
+
     @Override
     EurekaNamingService createServer(String clusterId, Supplier<String> serverAddressSupplier) throws Exception {
-        RestTemplateTransportClientFactory restTemplateTransportClientFactory =
-                new RestTemplateTransportClientFactory();
-        EurekaEndpoint eurekaEndpoint = new DefaultEndpoint(serverAddressSupplier.get());
+        RestTemplateTransportClientFactory restTemplateTransportClientFactory = new RestTemplateTransportClientFactory();
+        EurekaEndpoint eurekaEndpoint = new DefaultEndpoint(addHttpPrefix(serverAddressSupplier.get()));
         EurekaHttpClient eurekaHttpClient = restTemplateTransportClientFactory.newClient(eurekaEndpoint);
         return new EurekaNamingService(eurekaHttpClient);
     }
+
+    public String addHttpPrefix(String input) {
+        if (input == null || input.isEmpty()) {
+            return input;
+        }
+        if (!input.startsWith(HTTP_PREFIX) && !input.startsWith(HTTPS_PREFIX)) {
+            input = HTTP_PREFIX + input;
+        }
+
+        return input;
+    }
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/holder/NacosServerHolder.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/holder/NacosServerHolder.java
@@ -13,17 +13,23 @@
 package com.alibaba.nacossync.extension.holder;
 
 import com.alibaba.nacos.api.PropertyKeyConst;
+import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.NamingFactory;
 import com.alibaba.nacos.api.naming.NamingService;
+import com.alibaba.nacossync.constant.SkyWalkerConstants;
 import com.alibaba.nacossync.dao.ClusterAccessService;
+import com.alibaba.nacossync.dao.TaskAccessService;
 import com.alibaba.nacossync.pojo.model.ClusterDO;
+import com.alibaba.nacossync.pojo.model.TaskDO;
 import com.google.common.base.Joiner;
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.logging.log4j.util.Strings;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 /**
@@ -35,17 +41,30 @@
 public class NacosServerHolder extends AbstractServerHolderImpl<NamingService> {
 
     private final ClusterAccessService clusterAccessService;
+
+    private final TaskAccessService taskAccessService;
+
+    private static ConcurrentHashMap<String,NamingService> globalNameService = new ConcurrentHashMap<>(16);
 
-    public NacosServerHolder(ClusterAccessService clusterAccessService) {
+    public NacosServerHolder(ClusterAccessService clusterAccessService, TaskAccessService taskAccessService) {
         this.clusterAccessService = clusterAccessService;
+        this.taskAccessService = taskAccessService;
     }
 
     @Override
     NamingService createServer(String clusterId, Supplier<String> serverAddressSupplier)
         throws Exception {
+        String newClusterId;
+        if (clusterId.contains(":")) {
+            String[] split = clusterId.split(":");
+            newClusterId = split[1];
+        } else {
+            newClusterId = clusterId;
+        }
+        //代表此时为组合key，确定target集群中的nameService是不同的
         List<String> allClusterConnectKey = skyWalkerCacheServices
-            .getAllClusterConnectKey(clusterId);
-        ClusterDO clusterDO = clusterAccessService.findByClusterId(clusterId);
+            .getAllClusterConnectKey(newClusterId);
+        ClusterDO clusterDO = clusterAccessService.findByClusterId(newClusterId);
         String serverList = Joiner.on(",").join(allClusterConnectKey);
         Properties properties = new Properties();
         properties.setProperty(PropertyKeyConst.SERVER_ADDR, serverList);
@@ -58,6 +77,51 @@ NamingService createServer(String clusterId, Supplier<String> serverAddressSuppl
         Optional.ofNullable(clusterDO.getPassword()).ifPresent(value ->
             properties.setProperty(PropertyKeyConst.PASSWORD, value)
         );
-        return NamingFactory.createNamingService(properties);
+        NamingService namingService = NamingFactory.createNamingService(properties);
+        globalNameService.put(clusterId,namingService);
+        return namingService;
+    }
+
+    /**
+     * Get NamingService for different clients
+     * @param clusterId clusterId
+     * @return Returns Naming Service objects for different clusters
+     */
+    public NamingService getNameService(String clusterId){
+        return globalNameService.get(clusterId);
+    }
+
+    public NamingService getSourceNamingService(String taskId, String sourceClusterId) {
+        String key = taskId + sourceClusterId;
+        return serviceMap.computeIfAbsent(key, k->{
+            try {
+                log.info("Starting create source cluster server, key={}", key);
+                //代表此时为组合key，确定target集群中的nameService是不同的
+                List<String> allClusterConnectKey = skyWalkerCacheServices
+                        .getAllClusterConnectKey(sourceClusterId);
+                ClusterDO clusterDO = clusterAccessService.findByClusterId(sourceClusterId);
+                TaskDO task = taskAccessService.findByTaskId(taskId);
+                String serverList = Joiner.on(",").join(allClusterConnectKey);
+                Properties properties = new Properties();
+                properties.setProperty(PropertyKeyConst.SERVER_ADDR, serverList);
+                properties.setProperty(PropertyKeyConst.NAMESPACE, Optional.ofNullable(clusterDO.getNamespace()).orElse(
+                        Strings.EMPTY));
+                Optional.ofNullable(clusterDO.getUserName()).ifPresent(value ->
+                        properties.setProperty(PropertyKeyConst.USERNAME, value)
+                );
+
+                Optional.ofNullable(clusterDO.getPassword()).ifPresent(value ->
+                        properties.setProperty(PropertyKeyConst.PASSWORD, value)
+                );
+                properties.setProperty(SkyWalkerConstants.SOURCE_CLUSTERID_KEY,task.getSourceClusterId());
+                properties.setProperty(SkyWalkerConstants.DEST_CLUSTERID_KEY,task.getDestClusterId());
+                return NamingFactory.createNamingService(properties);
+            }catch (NacosException e) {
+                log.error("start source server fail,taskId:{},sourceClusterId:{}"
+                        , taskId, sourceClusterId, e);
+                return null;
+            }
+        });
+
     }
 }
diff --git a/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/impl/AbstractNacosSync.java b/nacossync-worker/src/main/java/com/alibaba/nacossync/extension/impl/AbstractNacosSync.java
@@ -101,7 +101,7 @@ public boolean delete(TaskDO taskDO) {
     }
 
     @Override
-    public boolean sync(TaskDO taskDO) {
+    public boolean sync(TaskDO taskDO, Integer index) {
         String taskId = taskDO.getTaskId();
         try {
             NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId());
@@ -128,7 +128,7 @@ public boolean sync(TaskDO taskDO) {
         return true;
     }
 
-    private void doSync(String taskId, TaskDO taskDO, NamingService sourceNamingService) throws NacosException {
+    private void doSync(String taskId, TaskDO taskDO, NamingService sourceNamingService) throws Exception {
         if (syncTaskTap.putIfAbsent(taskId, 1) != null) {
             log.info("任务Id:{}上一个同步任务尚未结束", taskId);
             return;
@@ -172,7 +172,7 @@ private void syncNewInstance(TaskDO taskDO, List<Instance> sourceInstances) thro
     }
 
 
-    private void removeInvalidInstance(TaskDO taskDO, List<Instance> sourceInstances) throws NacosException {
+    private void removeInvalidInstance(TaskDO taskDO, List<Instance> sourceInstances) throws Exception {
         String taskId = taskDO.getTaskId();
         if (this.sourceInstanceSnapshot.containsKey(taskId)) {
             Set<String> oldInstanceKeys = this.sourceInstanceSnapshot.get(taskId);
@@ -187,13 +187,23 @@ private void removeInvalidInstance(TaskDO taskDO, List<Instance> sourceInstances
         }
     }
 
+    @Override
+    public boolean needDelete(Map<String, String> destMetaData, TaskDO taskDO) {
+        return SyncService.super.needDelete(destMetaData, taskDO);
+    }
+
+    @Override
+    public boolean needSync(Map<String, String> sourceMetaData) {
+        return SyncService.super.needSync(sourceMetaData);
+    }
+
     public abstract String composeInstanceKey(String ip, int port);
 
     public abstract void register(TaskDO taskDO, Instance instance);
 
     public abstract void deregisterInstance(TaskDO taskDO) throws Exception;
 
-    public abstract void removeInvalidInstance(TaskDO taskDO, Set<String> invalidInstanceKeys);
+    public abstract void removeInvalidInstance(TaskDO taskDO, Set<String> invalidInstanceKeys) throws Exception;
 
     public NacosServerHolder getNacosServerHolder() {
         return nacosServerHolder;

diff --git a/...rker/src/main/java/com/alibaba/nacossync/extension/impl/ConsulSyncToNacosServiceImpl.java b/...rker/src/main/java/com/alibaba/nacossync/extension/impl/ConsulSyncToNacosServiceImpl.java
@@ -96,7 +96,7 @@ public boolean delete(TaskDO taskDO) {
     }
 
     @Override
-    public boolean sync(TaskDO taskDO) {
+    public boolean sync(TaskDO taskDO, Integer index) {
         try {
             ConsulClient consulClient = consulServerHolder.get(taskDO.getSourceClusterId());
             NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId());
@@ -106,7 +106,7 @@ public boolean sync(TaskDO taskDO) {
             Set<String> instanceKeys = new HashSet<>();
             overrideAllInstance(taskDO, destNamingService, healthServiceList, instanceKeys);
             cleanAllOldInstance(taskDO, destNamingService, instanceKeys);
-            specialSyncEventBus.subscribe(taskDO, this::sync);
+            specialSyncEventBus.subscribe(taskDO, t->sync(t, index));
         } catch (Exception e) {
             log.error("Sync task from consul to nacos was failed, taskId:{}", taskDO.getTaskId(), e);
             metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR);