feat(dlm):support viewing task logs

pom.xml

@@ -119,7 +119,7 @@
         <mina.version>2.1.6</mina.version>
 
         <!-- data-lifecycle-manager version -->
-        <data-lifecycle-manager.version>1.0.6</data-lifecycle-manager.version>
+        <data-lifecycle-manager.version>1.0.7</data-lifecycle-manager.version>
 
         <!-- plugin version -->
         <formatter-maven-plugin.version>2.11.0</formatter-maven-plugin.version>

server/integration-test/src/test/java/com/oceanbase/odc/metadb/schedule/ScheduleTaskRepositoryTest.java

@@ -16,6 +16,8 @@
 package com.oceanbase.odc.metadb.schedule;
 
 import java.util.Date;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Optional;
 
 import org.junit.Test;
@@ -60,6 +62,17 @@ public void updateExecutor() {
         Assert.equals(executor, byId.get().getExecutor());
     }
 
+    @Test
+    public void listByScheduleIdAndStatus() {
+        taskRepository.deleteAll();
+        createScheduleTask();
+        List<TaskStatus> statuses = new LinkedList<>();
+        statuses.add(TaskStatus.RUNNING);
+        statuses.add(TaskStatus.PREPARING);
+        List<ScheduleTaskEntity> byJobNameAndStatus = taskRepository.findByJobNameAndStatusIn("1", statuses);
+        Assert.equals(byJobNameAndStatus.size(), 1);
+    }
+
     private ScheduleTaskEntity createScheduleTask() {
         ScheduleTaskEntity entity = new ScheduleTaskEntity();
         entity.setStatus(TaskStatus.PREPARING);

server/odc-server/src/main/java/com/oceanbase/odc/server/web/controller/v2/ScheduleController.java

@@ -42,6 +42,7 @@
 import com.oceanbase.odc.service.schedule.model.ScheduleDetailResp;
 import com.oceanbase.odc.service.schedule.model.ScheduleStatus;
 import com.oceanbase.odc.service.schedule.model.ScheduleTaskResp;
+import com.oceanbase.odc.service.task.model.OdcTaskLogLevel;
 
 import io.swagger.annotations.ApiOperation;
 
@@ -140,4 +141,12 @@ public SuccessResponse<ScheduleTaskResp> rollbackTask(@PathVariable Long schedul
         return Responses.single(scheduleService.rollbackTask(scheduleId, taskId));
     }
 
+
+    @ApiOperation(value = "GetScheduleTaskLog", notes = "获取计划任务日志")
+    @RequestMapping(value = "/schedules/{scheduleId:[\\d]+}/tasks/{taskId:[\\d]+}/log", method = RequestMethod.GET)
+    public SuccessResponse<String> getScheduleTaskLog(@PathVariable Long scheduleId, @PathVariable Long taskId,
+            @RequestParam OdcTaskLogLevel logType) {
+        return Responses.single(scheduleService.getLog(scheduleId, taskId, logType));
+    }
+
 }

server/odc-server/src/main/resources/data.sql

@@ -684,6 +684,9 @@ INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES
 INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.dlm.single-task-thread-pool-size', '12', 'DLM 单个任务可用线程数' ) ON DUPLICATE KEY UPDATE `id` = `id`;
 INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.dlm.thread-pool-size', '100', '单个 POD 中 DLM 任务线程池大小' ) ON DUPLICATE KEY UPDATE `id` = `id`;
 INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.dlm.support-breakpoint-recovery', 'true', 'DLM 任务是否开启断点恢复' ) ON DUPLICATE KEY UPDATE `id` = `id`;
+INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.dlm.task-connection-query-timeout-seconds', '180', 'DLM 任务 SQL 超时时间' ) ON DUPLICATE KEY UPDATE `id` = `id`;
+INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.dlm.sharding-strategy', 'FIXED_LENGTH', 'DLM 分片策略，默认值 FIXED_LENGTH 表示均匀分片，适合小规格实例。使用 MATCH 策略时将出现少量慢 SQL，整体性能会有较大提升，适合大规格实例。' ) ON DUPLICATE KEY UPDATE `id` = `id`;
+INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.dlm.default-scan-batch-size', '10000', 'DLM 分片大小，默认值 10000 表示分片 SQL 每次会扫描 10000 个主键' ) ON DUPLICATE KEY UPDATE `id` = `id`;
 
 INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.datatransfer.use-server-prep-stmts', 'true', '导入导出是否开启 ps 协议，默认为开启' ) ON DUPLICATE KEY UPDATE `id` = `id`;
 INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES( 'odc.task.datatransfer.cursor-fetch-size', '20', '导出时游标的 fetch size，默认为 20，最大值为 1000' ) ON DUPLICATE KEY UPDATE `id` = `id`;

server/odc-server/src/main/resources/log4j2.xml

@@ -488,6 +488,51 @@
             </Routes>
         </Routing>
 
+        <Routing name="ScheduleTaskInfoRoutingAppender">
+            <Routes pattern="${ctx:scheduleTaskId}">
+                <Route>
+                    <RollingFile name="InfoRolling-${ctx:scheduleTaskId}"
+                                 fileName="${LOG_DIRECTORY}/scheduleTask/${ctx:jobName}-${ctx:jobGroup}/${ctx:scheduleTaskId}/log.all"
+                                 filePattern="${LOG_DIRECTORY}/scheduleTask/${ctx:jobName}-${ctx:jobGroup}/${ctx:scheduleTaskId}/${date:yyyy-MM}/log-%d{yyyy-MM-dd}-%i.all.gz">
+                        <PatternLayout>
+                            <pattern>%d{yyyy-MM-dd HH:mm:ss} [%p] %m%n</pattern>
+                        </PatternLayout>
+                        <Filters>
+                            <!-- TRACE < DEBUG < INFO < WARN < ERROR < FATAL -->
+                            <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
+                        </Filters>
+                        <Policies>
+                            <TimeBasedTriggeringPolicy interval="6" modulate="true"/>
+                            <SizeBasedTriggeringPolicy size="256 MB"/>
+                        </Policies>
+                    </RollingFile>
+                </Route>
+            </Routes>
+        </Routing>
+
+        <Routing name="ScheduleTaskWarnRoutingAppender">
+            <Routes pattern="${ctx:scheduleTaskId}">
+                <Route>
+                    <RollingFile name="WarnRolling-${ctx:scheduleTaskId}"
+                                 fileName="${LOG_DIRECTORY}/scheduleTask/${ctx:jobName}-${ctx:jobGroup}/${ctx:scheduleTaskId}/log.warn"
+                                 filePattern="${LOG_DIRECTORY}/scheduleTask/${ctx:jobName}-${ctx:jobGroup}/${ctx:scheduleTaskId}/${date:yyyy-MM}/log-%d{yyyy-MM-dd}-%i.warn.gz">
+                        <PatternLayout>
+                            <pattern>%d{yyyy-MM-dd HH:mm:ss} [%p] %m%n</pattern>
+                        </PatternLayout>
+                        <Filters>
+                            <!-- TRACE < DEBUG < INFO < WARN < ERROR < FATAL -->
+                            <ThresholdFilter level="ERROR" onMatch="DENY" onMismatch="NEUTRAL"/>
+                            <ThresholdFilter level="WARN" onMatch="ACCEPT" onMismatch="DENY"/>
+                        </Filters>
+                        <Policies>
+                            <TimeBasedTriggeringPolicy interval="6" modulate="true"/>
+                            <SizeBasedTriggeringPolicy size="256 MB"/>
+                        </Policies>
+                    </RollingFile>
+                </Route>
+            </Routes>
+        </Routing>
+
 
     </Appenders>
     <Loggers>
@@ -588,8 +633,9 @@
             <AppenderRef ref="WarnRoutingOnlineSchemaChangeAppender"/>
         </Logger>
 
-        <Logger name="com.oceanbase.tools.migrator" level="INFO">
-            <AppenderRef ref="OdcFileAppender"/>
+        <Logger name="com.oceanbase.tools.migrator" level="INFO" additivity="true">
+            <AppenderRef ref="ScheduleTaskInfoRoutingAppender"/>
+            <AppenderRef ref="ScheduleTaskWarnRoutingAppender"/>
         </Logger>
 
         <!-- ODC程序日志输出，输出级别 INFO -->

server/odc-service/src/main/java/com/oceanbase/odc/metadb/schedule/ScheduleTaskRepository.java

@@ -15,6 +15,8 @@
  */
 package com.oceanbase.odc.metadb.schedule;
 
+import java.util.List;
+
 import javax.transaction.Transactional;
 
 import org.springframework.data.jpa.repository.JpaRepository;
@@ -35,6 +37,8 @@
 public interface ScheduleTaskRepository extends JpaRepository<ScheduleTaskEntity, Long>,
         JpaSpecificationExecutor<ScheduleTaskEntity> {
 
+    List<ScheduleTaskEntity> findByJobNameAndStatusIn(String jobName, List<TaskStatus> statuses);
+
     @Transactional
     @Modifying
     @Query("update ScheduleTaskEntity st set st.status = ?2 where st.id = ?1")

server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/DLMConfiguration.java

@@ -21,6 +21,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import com.oceanbase.tools.migrator.common.enums.ShardingStrategy;
 import com.oceanbase.tools.migrator.core.IJobStore;
 
 import lombok.extern.slf4j.Slf4j;
@@ -44,13 +45,25 @@ public class DLMConfiguration {
     @Value("${odc.task.dlm.single-task-thread-pool-size:15}")
     private int singleTaskThreadPoolSize;
 
+    @Value("${odc.task.dlm.task-connection-query-timeout-seconds:180}")
+    private int taskConnectionQueryTimeout;
+
+    @Value("${odc.task.dlm.sharding-strategy:FIXED_LENGTH}")
+    private ShardingStrategy shardingStrategy;
+
+    @Value("${odc.task.dlm.default-scan-batch-size:10000}")
+    private int defaultScanBatchSize;
+
     @Bean
     public JobMetaFactory jobMetaFactory(IJobStore jobStore) {
         JobMetaFactory jobMetaFactory = new JobMetaFactory();
         jobMetaFactory.setJobStore(jobStore);
         jobMetaFactory.setExecutorService(Executors.newFixedThreadPool(dlmThreadPoolSize));
         jobMetaFactory.setSingleTaskThreadPoolSize(singleTaskThreadPoolSize);
         jobMetaFactory.setReadWriteRatio(readWriteRatio);
+        jobMetaFactory.setTaskConnectionQueryTimeout(taskConnectionQueryTimeout);
+        jobMetaFactory.setDefaultShardingStrategy(shardingStrategy);
+        jobMetaFactory.setDefaultScanBatchSize(defaultScanBatchSize);
         return jobMetaFactory;
     }
 

server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/DataSourceInfoBuilder.java

@@ -55,7 +55,7 @@ public static DataSourceInfo build(ConnectionConfig connectionConfig) {
                         .setFullUserName(OBConsoleDataSourceFactory.getUsername(connectionConfig));
                 dataSourceInfo.setDbType(DataBaseType.OCEANBASEV10.name());
                 dataSourceInfo.setSysUser(connectionConfig.getSysTenantUsername());
-                dataSourceInfo.setUserLocalProxy(false);
+                dataSourceInfo.setClusterName(connectionConfig.getClusterName());
                 if (StringUtils.isNotEmpty(connectionConfig.getSysTenantPassword())) {
                     try {
                         dataSourceInfo.setSysPassword(EncryptUtils.encode(connectionConfig.getSysTenantPassword()));

server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/JobMetaFactory.java

@@ -23,6 +23,7 @@
 import com.oceanbase.tools.migrator.common.dto.HistoryJob;
 import com.oceanbase.tools.migrator.common.enums.JobStatus;
 import com.oceanbase.tools.migrator.common.enums.JobType;
+import com.oceanbase.tools.migrator.common.enums.ShardingStrategy;
 import com.oceanbase.tools.migrator.core.AbstractJobMetaFactory;
 import com.oceanbase.tools.migrator.core.JobReq;
 import com.oceanbase.tools.migrator.core.meta.ClusterMeta;
@@ -41,8 +42,14 @@ public class JobMetaFactory extends AbstractJobMetaFactory {
 
     private final SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
     private int singleTaskThreadPoolSize;
+
+    private int taskConnectionQueryTimeout;
     private double readWriteRatio;
 
+    private int defaultScanBatchSize;
+
+    private ShardingStrategy defaultShardingStrategy;
+
     public JobMeta create(DlmTask parameters) throws Exception {
         HistoryJob historyJob = new HistoryJob();
         historyJob.setId(parameters.getId());
@@ -58,18 +65,23 @@ public JobMeta create(DlmTask parameters) throws Exception {
         LogicTableConfig logicTableConfig = parameters.getLogicTableConfig();
         logicTableConfig.setReaderTaskCount((int) (singleTaskThreadPoolSize * readWriteRatio / (1 + readWriteRatio)));
         logicTableConfig.setWriterTaskCount(singleTaskThreadPoolSize - logicTableConfig.getReaderTaskCount());
+        logicTableConfig.setGeneratorBatchSize(defaultScanBatchSize);
         DataSourceInfo sourceInfo = DataSourceInfoBuilder.build(parameters.getSourceDs());
         DataSourceInfo targetInfo = DataSourceInfoBuilder.build(parameters.getTargetDs());
         sourceInfo.setConnectionCount(2 * (logicTableConfig.getReaderTaskCount()
                 + parameters.getLogicTableConfig().getWriterTaskCount()));
         targetInfo.setConnectionCount(2 * (logicTableConfig.getReaderTaskCount()
                 + parameters.getLogicTableConfig().getWriterTaskCount()));
+        sourceInfo.setQueryTimeout(taskConnectionQueryTimeout);
+        targetInfo.setQueryTimeout(taskConnectionQueryTimeout);
         log.info("Begin to create dlm job,params={}", logicTableConfig);
         // ClusterMeta and TenantMeta used to calculate min limit size.
         JobReq req =
                 new JobReq(historyJob, parameters.getLogicTableConfig(), sourceInfo, targetInfo, new ClusterMeta(),
                         new ClusterMeta(), new TenantMeta(), new TenantMeta());
-        return super.create(req);
+        JobMeta jobMeta = super.create(req);
+        jobMeta.setShardingStrategy(defaultShardingStrategy);
+        return jobMeta;
     }
 
     public void setReadWriteRatio(double readWriteRatio) {
@@ -80,4 +92,16 @@ public void setSingleTaskThreadPoolSize(int singleTaskThreadPoolSize) {
         this.singleTaskThreadPoolSize = singleTaskThreadPoolSize;
     }
 
+    public void setTaskConnectionQueryTimeout(int taskConnectionQueryTimeout) {
+        this.taskConnectionQueryTimeout = taskConnectionQueryTimeout;
+    }
+
+    public void setDefaultShardingStrategy(ShardingStrategy defaultShardingStrategy) {
+        this.defaultShardingStrategy = defaultShardingStrategy;
+    }
+
+    public void setDefaultScanBatchSize(int defaultScanBatchSize) {
+        this.defaultScanBatchSize = defaultScanBatchSize;
+    }
+
 }

server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/utils/DataArchiveConditionUtil.java

@@ -19,13 +19,16 @@
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.apache.commons.collections4.CollectionUtils;
+
 import com.oceanbase.odc.service.dlm.model.OffsetConfig;
 
 /**
@@ -58,6 +61,9 @@ private static String replaceVariables(String condition, Map<String, String> var
     }
 
     private static Map<String, String> getVariablesMap(List<OffsetConfig> variables) {
+        if (CollectionUtils.isEmpty(variables)) {
+            return Collections.emptyMap();
+        }
         Map<String, String> map = new HashMap<>();
         variables.forEach(obj -> {
             if (map.containsKey(obj.getName())) {

server/odc-service/src/main/java/com/oceanbase/odc/service/quartz/executor/AbstractQuartzJob.java

@@ -50,10 +50,13 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
         log.info("Start to run new job,jobKey={}", context.getJobDetail().getKey());
         try {
             odcJob = getOdcJob(context);
+            odcJob.before(context);
             run(context);
         } catch (Exception e) {
             log.warn("Job execute failed,job key={},fire time={}.",
                     context.getJobDetail().getKey(), context.getFireTime(), e);
+        } finally {
+            odcJob.after(context);
         }
 
         log.info("Job done,jobKey={}", context.getJobDetail().getKey());