数据库快照功能BUG修复

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/DatacenterApplication.java

@@ -7,6 +7,7 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
 import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.scheduling.annotation.EnableAsync;
 
 /**
  * @author yingp
@@ -20,6 +21,7 @@ import org.springframework.cloud.openfeign.EnableFeignClients;
 @EnableFeignClients({
         "com.usoftchina.saas.account.api"
 })
+@EnableAsync
 public class DatacenterApplication {
 
     public static void main(String[] args) {

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/controller/SnapshotController.java

@@ -4,6 +4,7 @@ import com.github.pagehelper.PageInfo;
 import com.usoftchina.saas.base.Result;
 import com.usoftchina.saas.dc.po.Snapshot;
 import com.usoftchina.saas.dc.service.SnapshotService;
+import com.usoftchina.saas.dc.service.SnapshotUsageService;
 import com.usoftchina.saas.page.PageDefault;
 import com.usoftchina.saas.page.PageRequest;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -20,6 +21,9 @@ public class SnapshotController {
     @Autowired
     private SnapshotService snapshotService;
 
+    @Autowired
+    private SnapshotUsageService snapshotUsageService;
+
     /**
      * 生成快照
      *
@@ -60,7 +64,8 @@ public class SnapshotController {
      * @return
      */
     @PostMapping("/restore")
-    public Result restore(@RequestParam Long id) {
+    public Result restore(@RequestParam String id) {
+        snapshotUsageService.save(id);
         return Result.success();
     }
 }

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/mapper/CommonMapper.java

@@ -4,6 +4,7 @@ import org.apache.ibatis.annotations.Param;
 
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * @author yingp
@@ -21,4 +22,36 @@ public interface CommonMapper {
     List<LinkedHashMap<String, Object>> selectByTableAndColumnAndValue(
             @Param("tableName") String tableName, @Param("columnName") String columnName,
             @Param("columnValue") Object columnValue);
+
+    /**
+     * 按指定表、字段、值删除
+     *
+     * @param tableName
+     * @param columnName
+     * @param columnValue
+     * @return
+     */
+    int deleteByTableAndColumnAndValue(
+            @Param("tableName") String tableName, @Param("columnName") String columnName,
+            @Param("columnValue") Object columnValue);
+
+    /**
+     * 插入map数据
+     *
+     * @param tableName
+     * @param record
+     * @return
+     */
+    int insertMap(@Param("tableName") String tableName, @Param("record") Map<String, Object> record);
+
+    /**
+     * 插入list数据
+     *
+     * @param tableName
+     * @param columns
+     * @param data
+     * @return
+     */
+    int insertListMap(@Param("tableName") String tableName, @Param("columns") List<String> columns,
+                      @Param("data") List<Map<String, Object>> data);
 }

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/po/SnapshotUsage.java

@@ -1,5 +1,6 @@
 package com.usoftchina.saas.dc.po;
 
+import com.usoftchina.saas.context.BaseContextHolder;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.index.Indexed;
 import org.springframework.data.mongodb.core.mapping.Document;
@@ -39,6 +40,18 @@ public class SnapshotUsage implements Serializable {
     private short status;
     private String message;
 
+    public SnapshotUsage() {
+        this.createTime = System.currentTimeMillis();
+        this.creatorId = BaseContextHolder.getUserId();
+        this.companyId = BaseContextHolder.getCompanyId();
+        this.status = SnapshotUsage.Status.NOT_READY;
+    }
+
+    public SnapshotUsage(String usedSnapshotId) {
+        this();
+        this.usedSnapshotId = usedSnapshotId;
+    }
+
     public String get_id() {
         return _id;
     }
@@ -106,22 +119,14 @@ public class SnapshotUsage implements Serializable {
     public interface Status {
         short NOT_READY = 0;
         /**
-         * 导出当前数据
-         */
-        short DUMPING_ORIGIN = 1;
-        /**
-         * 读取备份文件
-         */
-        short READ_DUMPFILE = 1;
-        /**
-         * 还原中(删除旧数据，)
+         * 还原中(删除当前数据，使用快照数据写入)
          */
         short RESTORE = 1;
         /**
          * 操作失败需要使用originData还原
          */
         short ROLLBACK = -1;
-        short FAILED = 2;
+        short FAILED = -2;
         short SUCCESS = 2;
     }
 }

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/repository/SnapshotDataRepository.java

@@ -4,6 +4,8 @@ import com.usoftchina.saas.dc.po.SnapshotData;
 import org.springframework.data.mongodb.repository.MongoRepository;
 import org.springframework.stereotype.Repository;
 
+import java.util.List;
+
 /**
  * @author yingp
  * @date 2019/1/22
@@ -16,4 +18,13 @@ public interface SnapshotDataRepository extends MongoRepository<SnapshotData, St
      * @param snapshotId
      */
     void deleteBySnapshotId(String snapshotId);
+
+    /**
+     * 查找
+     * @param snapshotId
+     * @param dcName
+     * @param dbName
+     * @return
+     */
+    List<SnapshotData> findBySnapshotIdAndDcNameAndDbName(String snapshotId, String dcName, String dbName);
 }

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/service/SnapshotUsageService.java

@@ -0,0 +1,23 @@
+package com.usoftchina.saas.dc.service;
+
+import com.usoftchina.saas.dc.po.SnapshotUsage;
+
+/**
+ * @author yingp
+ * @date 2019/1/23
+ */
+public interface SnapshotUsageService {
+    /**
+     * 保存
+     * @param snapshotId
+     * @return
+     */
+    SnapshotUsage save(String snapshotId);
+
+    /**
+     * 查找
+     * @param id
+     * @return
+     */
+    SnapshotUsage selectByPrimaryKey(String id);
+}

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/service/impl/SnapshotUsageServiceImpl.java

@@ -0,0 +1,45 @@
+package com.usoftchina.saas.dc.service.impl;
+
+import com.usoftchina.saas.context.SpringContextHolder;
+import com.usoftchina.saas.dc.po.Snapshot;
+import com.usoftchina.saas.dc.po.SnapshotUsage;
+import com.usoftchina.saas.dc.repository.SnapshotRepository;
+import com.usoftchina.saas.dc.repository.SnapshotUsageRepository;
+import com.usoftchina.saas.dc.service.SnapshotUsageService;
+import com.usoftchina.saas.dc.snapshot.event.ReadyToCreateEvent;
+import com.usoftchina.saas.dc.snapshot.event.ReadyToRestoreEvent;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.Optional;
+
+/**
+ * @author yingp
+ * @date 2019/1/23
+ */
+@Service
+public class SnapshotUsageServiceImpl implements SnapshotUsageService{
+
+    @Autowired
+    private SnapshotUsageRepository snapshotUsageRepository;
+
+    @Autowired
+    private SnapshotRepository snapshotRepository;
+
+    @Override
+    public SnapshotUsage save(String snapshotId) {
+        Optional<Snapshot> optional = snapshotRepository.findById(snapshotId);
+        if (optional.isPresent()) {
+            SnapshotUsage usage = snapshotUsageRepository.save(new SnapshotUsage(snapshotId));
+            SpringContextHolder.getContext().publishEvent(new ReadyToRestoreEvent(this,
+                    optional.get(), usage));
+            return usage;
+        }
+        return null;
+    }
+
+    @Override
+    public SnapshotUsage selectByPrimaryKey(String id) {
+        return snapshotUsageRepository.findById(id).orElse(null);
+    }
+}

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/snapshot/event/ReadyToRestoreEvent.java

@@ -0,0 +1,15 @@
+package com.usoftchina.saas.dc.snapshot.event;
+
+import com.usoftchina.saas.dc.po.Snapshot;
+import com.usoftchina.saas.dc.po.SnapshotUsage;
+
+/**
+ * @author yingp
+ * @date 2019/1/23
+ */
+public class ReadyToRestoreEvent extends SnapshotUseEvent {
+
+    public ReadyToRestoreEvent(Object source, Snapshot snapshot, SnapshotUsage usage) {
+        super(source, snapshot, usage);
+    }
+}

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/snapshot/event/RestoreFailedEvent.java

@@ -0,0 +1,22 @@
+package com.usoftchina.saas.dc.snapshot.event;
+
+import com.usoftchina.saas.dc.po.Snapshot;
+import com.usoftchina.saas.dc.po.SnapshotUsage;
+
+/**
+ * @author yingp
+ * @date 2019/1/22
+ */
+public class RestoreFailedEvent extends SnapshotUseEvent{
+
+    private Exception exception;
+
+    public RestoreFailedEvent(Object source, Snapshot snapshot, SnapshotUsage usage, Exception exception) {
+        super(source, snapshot, usage);
+        this.exception = exception;
+    }
+
+    public Exception getException() {
+        return exception;
+    }
+}

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/snapshot/event/RestoredEvent.java

@@ -0,0 +1,15 @@
+package com.usoftchina.saas.dc.snapshot.event;
+
+import com.usoftchina.saas.dc.po.Snapshot;
+import com.usoftchina.saas.dc.po.SnapshotUsage;
+
+/**
+ * @author yingp
+ * @date 2019/1/23
+ */
+public class RestoredEvent extends SnapshotUseEvent {
+
+    public RestoredEvent(Object source, Snapshot snapshot, SnapshotUsage usage) {
+        super(source, snapshot, usage);
+    }
+}

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/snapshot/event/SnapshotLifecycle.java

@@ -5,6 +5,7 @@ import com.usoftchina.saas.account.cache.CompanyLockCache;
 import com.usoftchina.saas.context.SpringContextHolder;
 import com.usoftchina.saas.dc.po.DataSourceInfo;
 import com.usoftchina.saas.dc.po.Snapshot;
+import com.usoftchina.saas.dc.po.SnapshotUsage;
 import com.usoftchina.saas.dc.repository.SnapshotDataRepository;
 import com.usoftchina.saas.dc.repository.SnapshotRepository;
 import com.usoftchina.saas.dc.repository.SnapshotUsageRepository;
@@ -92,7 +93,6 @@ public class SnapshotLifecycle {
     @EventListener(CreateFailedEvent.class)
     public void onCreateFailedEvent(CreateFailedEvent event) {
         Snapshot snapshot = event.getSnapshot();
-        event.getException().printStackTrace();
         logger.error("failed to create snapshot {}", event.getException(), snapshot.get_id());
         // 解锁
         CompanyLockCache.unlock(snapshot.getCompanyId());
@@ -111,4 +111,63 @@ public class SnapshotLifecycle {
         snapshotUsageRepository.deleteByUsedSnapshotId(snapshot.get_id());
         snapshotUsageRepository.deleteByOriginSnapshotId(snapshot.get_id());
     }
+
+    @Async
+    @EventListener(ReadyToRestoreEvent.class)
+    public void onReadyToRestoreEvent(ReadyToRestoreEvent event) {
+        Snapshot snapshot = event.getSnapshot();
+        SnapshotUsage usage = event.getUsage();
+        try {
+            usage.setStatus(SnapshotUsage.Status.RESTORE);
+            snapshotUsageRepository.save(usage);
+            // 为防止导入过程中有人操作，导致部分数据不一致，在导出前锁定该公司不允许操作
+            CompanyLockCache.lock(usage.getCompanyId());
+            restoreSnapshotData(snapshot);
+            SpringContextHolder.getContext().publishEvent(new RestoredEvent(this, snapshot, usage));
+        } catch (Exception e) {
+            SpringContextHolder.getContext().publishEvent(new RestoreFailedEvent(this, snapshot, usage, e));
+        }
+    }
+
+    /**
+     * 连接到不同数据中心，查找数据，生成快照数据，删除，使用快照数据写入
+     *
+     * @param snapshot
+     */
+    private void restoreSnapshotData(Snapshot snapshot) {
+        String dcName = StringUtils.nullIf(CompanyCache.current().getCompany().getDcName(), "default");
+        List<DataSourceInfo> dss = dataSourceInfoService.findByDcNameUseDefault(dcName);
+        if (!CollectionUtils.isEmpty(dss)) {
+            dss.parallelStream().forEach(ds -> {
+                if ("mysql".equals(ds.getDbType())) {
+                    mysqlStrategy.imp(snapshot, ds);
+                }
+            });
+        }
+    }
+
+    @Async
+    @EventListener(RestoredEvent.class)
+    public void onRestoredEvent(RestoredEvent event) {
+        SnapshotUsage usage = event.getUsage();
+        // 解锁
+        CompanyLockCache.unlock(usage.getCompanyId());
+        usage.setStatus(SnapshotUsage.Status.SUCCESS);
+        snapshotUsageRepository.save(usage);
+    }
+
+    @Async
+    @EventListener(RestoreFailedEvent.class)
+    public void onRestoreFailedEvent(RestoreFailedEvent event) {
+        Snapshot snapshot = event.getSnapshot();
+        SnapshotUsage usage = event.getUsage();
+        logger.error("failed to restore by snapshot {}", event.getException(), snapshot.get_id());
+        // 解锁
+        CompanyLockCache.unlock(usage.getCompanyId());
+        // 回滚
+        usage.setStatus(SnapshotUsage.Status.FAILED);
+        usage.setMessage(event.getException().getMessage());
+        snapshotUsageRepository.save(usage);
+        // TODO
+    }
 }

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/snapshot/event/SnapshotUseEvent.java

@@ -0,0 +1,22 @@
+package com.usoftchina.saas.dc.snapshot.event;
+
+import com.usoftchina.saas.dc.po.Snapshot;
+import com.usoftchina.saas.dc.po.SnapshotUsage;
+
+/**
+ * @author yingp
+ * @date 2019/1/23
+ */
+public abstract class SnapshotUseEvent extends SnapshotEvent {
+
+    private SnapshotUsage usage;
+
+    public SnapshotUseEvent(Object source, Snapshot snapshot, SnapshotUsage usage) {
+        super(source, snapshot);
+        this.usage = usage;
+    }
+
+    public SnapshotUsage getUsage() {
+        return usage;
+    }
+}

base-servers/datacenter/datacenter-server/src/main/java/com/usoftchina/saas/dc/snapshot/support/MysqlStrategy.java

@@ -13,8 +13,7 @@ import com.usoftchina.saas.utils.CollectionUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-import java.util.LinkedHashMap;
-import java.util.List;
+import java.util.*;
 
 /**
  * @author yingp
@@ -63,4 +62,30 @@ public class MysqlStrategy implements CompanyStrategy {
             DynamicDataSourceContextHolder.clear();
         }
     }
+
+    public void imp(final Snapshot snapshot, final DataSourceInfo ds) {
+        List<SnapshotData> snapshotDataList = snapshotDataRepository.findBySnapshotIdAndDcNameAndDbName(
+                snapshot.get_id(), ds.getDcName(), ds.getDbName());
+        if (!CollectionUtils.isEmpty(snapshotDataList)) {
+            final long companyId = snapshot.getCompanyId();
+            // 如果不存在数据源则自动创建
+            dataSourceRegister.createDataSource(ds);
+            snapshotDataList.parallelStream().forEach(snapshotData -> {
+                try {
+                    DynamicDataSourceContextHolder.set(ds);
+                    // 删除当前数据
+                    commonMapper.deleteByTableAndColumnAndValue(snapshotData.getTableName(),
+                            snapshotData.getColumnName(), companyId);
+                    // 写入快照数据
+                    List<Map<String, Object>> list = (List<Map<String, Object>>) snapshotData.getData();
+                    if (!CollectionUtils.isEmpty(list)) {
+                        commonMapper.insertListMap(snapshotData.getTableName(),
+                                new ArrayList<>(list.get(0).keySet()), list);
+                    }
+                } finally {
+                    DynamicDataSourceContextHolder.clear();
+                }
+            });
+        }
+    }
 }

base-servers/datacenter/datacenter-server/src/main/resources/mapper/CommonMapper.xml

@@ -4,4 +4,29 @@
     <select id="selectByTableAndColumnAndValue" resultType="java.util.LinkedHashMap" statementType="STATEMENT">
         select * from ${tableName} where ${columnName}=${columnValue}
     </select>
+    <delete id="deleteByTableAndColumnAndValue" statementType="STATEMENT">
+        delete from ${tableName} where ${columnName}=${columnValue}
+    </delete>
+    <insert id="insertMap" statementType="STATEMENT">
+        insert into ${tableName}
+        <foreach collection="record.keys" item="key" open="(" close=")" separator="," >
+            ${key}
+        </foreach>
+        values
+        <foreach collection="record.keys" item="key" open="(" close=")" separator=",">
+            #{record[${key}]}
+        </foreach>
+    </insert>
+    <insert id="insertListMap">
+        insert into ${tableName}
+        <foreach collection="columns" item="key" open="(" close=")" separator="," >
+            ${key}
+        </foreach>
+        values
+        <foreach collection="data" item="record" index="index" separator =",">
+            <foreach collection="columns" item="key" index="keyIndex" open="(" close=")" separator=",">
+                #{data[${index}].${key}}
+            </foreach>
+        </foreach>
+    </insert>
 </mapper>

base-servers/datacenter/datacenter-server/src/main/resources/mapper/SchemaMapper.xml

@@ -6,10 +6,11 @@
         <result column="column_name" jdbcType="VARCHAR" property="columnName"/>
     </resultMap>
     <select id="selectSchemaInfoByColumns" resultMap="BaseResultMap">
-        select table_name,column_name from information_schema.`COLUMNS` where table_schema=#{schema,jdbcType=VARCHAR}
-        and column_name in
+        select c.table_name,c.column_name from information_schema.`COLUMNS` c,information_schema.`TABLES` t where
+        t.table_schema=#{schema,jdbcType=VARCHAR} and t.table_type='BASE TABLE' and c.table_schema=t.table_schema
+        and c.table_name=t.table_name and c.column_name in
         <foreach collection="columnNames" index="index" item="columnName" open="(" separator="," close=")">
-          #{columnName}
+            #{columnName}
         </foreach>
     </select>
 </mapper>

base-servers/datacenter/datacenter-server/src/test/java/com/usoftchina/saas/dc/SnapshotServiceTest.java

@@ -35,11 +35,11 @@ public class SnapshotServiceTest {
         while(true) {
             snapshot = snapshotService.selectByPrimaryKey(snapshot.get_id());
             System.out.println("Status: " + snapshot.getStatus());
+            Thread.sleep(1000);
             if (snapshot.getStatus() == Snapshot.Status.SUCCESS ||
                     snapshot.getStatus() == Snapshot.Status.FAILED) {
                 break;
             }
-            Thread.sleep(1000);
         }
     }
 }

base-servers/datacenter/datacenter-server/src/test/java/com/usoftchina/saas/dc/SnapshotUsageServiceTest.java

@@ -0,0 +1,45 @@
+package com.usoftchina.saas.dc;
+
+import com.usoftchina.saas.context.BaseContextHolder;
+import com.usoftchina.saas.dc.po.SnapshotUsage;
+import com.usoftchina.saas.dc.service.SnapshotUsageService;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.MethodSorters;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author yingp
+ * @date 2019/1/22
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class SnapshotUsageServiceTest {
+
+    @Autowired
+    private SnapshotUsageService snapshotUsageService;
+
+    @Test
+    public void testA_save() throws Exception {
+        BaseContextHolder.setCompanyId(134);
+        BaseContextHolder.setUserId(41);
+        SnapshotUsage usage = snapshotUsageService.save("5c47b8bcae463e94f090f7a3");
+        printStatus(usage);
+    }
+
+    private void printStatus(SnapshotUsage usage) throws Exception {
+        while(true) {
+            usage = snapshotUsageService.selectByPrimaryKey(usage.get_id());
+            System.out.println("Status: " + usage.getStatus());
+            Thread.sleep(1000);
+            if (usage.getStatus() == SnapshotUsage.Status.SUCCESS ||
+                    usage.getStatus() == SnapshotUsage.Status.FAILED) {
+                break;
+            }
+        }
+    }
+}