DM/Openapi: support start task by some conditions

dm/dm/config/task_converters.go

@@ -166,10 +166,7 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *DBConfig, sourceCf
 			}
 			subTaskCfg.Meta = meta
 		}
-		// check must set meta when mode is ModeIncrement
-		if subTaskCfg.Meta == nil && subTaskCfg.Mode == ModeIncrement {
-			return nil, terror.ErrConfigMetadataNotSet.Generate(i, ModeIncrement)
-		}
+
 		// set shard config
 		if task.ShardMode != nil {
 			subTaskCfg.IsSharding = true
@@ -684,3 +681,18 @@ func genFilterRuleName(sourceName string, idx int) string {
 	// NOTE that we don't have user input filter rule name in sub task config, so we make one by ourself
 	return fmt.Sprintf("%s-filter-rule-%d", sourceName, idx)
 }
+
+func OpenAPIStartTaskReqToTaskCliArgs(req openapi.StartTaskRequest) (*TaskCliArgs, error) {
+	cliArgs := &TaskCliArgs{}
+	if req.StartTime != nil {
+		cliArgs.StartTime = *req.StartTime
+	}
+	if req.SafeModeTimeDuration != nil {
+		cliArgs.SafeModeDuration = *req.SafeModeTimeDuration
+	}
+
+	if err := cliArgs.Verify(); err != nil {
+		return nil, err
+	}
+	return cliArgs, nil
+}

dm/dm/master/openapi_controller.go

@@ -23,6 +23,9 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/pingcap/tiflow/dm/dm/master/scheduler"
+	"github.com/pingcap/tiflow/dm/pkg/ha"
+
 	"github.com/pingcap/tiflow/dm/checker"
 	dmcommon "github.com/pingcap/tiflow/dm/dm/common"
 	"github.com/pingcap/tiflow/dm/dm/config"
@@ -652,10 +655,14 @@ func (s *Server) startTask(ctx context.Context, taskName string, req openapi.Sta
 		return nil
 	}
 
-	// TODO(ehco) support other start args after https://github.com/pingcap/tiflow/pull/4601 merged
+	var (
+		release scheduler.ReleaseFunc
+		err     error
+	)
+	// removeMeta
 	if req.RemoveMeta != nil && *req.RemoveMeta {
 		// use same latch for remove-meta and start-task
-		release, err := s.scheduler.AcquireSubtaskLatch(taskName)
+		release, err = s.scheduler.AcquireSubtaskLatch(taskName)
 		if err != nil {
 			return terror.ErrSchedulerLatchInUse.Generate("RemoveMeta", taskName)
 		}
@@ -666,8 +673,28 @@ func (s *Server) startTask(ctx context.Context, taskName string, req openapi.Sta
 		if err != nil {
 			return terror.Annotate(err, "while removing metadata")
 		}
+	}
+
+	cliArgs, err := config.OpenAPIStartTaskReqToTaskCliArgs(req)
+	if err != nil {
+		return terror.Annotate(err, "while converting task command line arguments")
+	}
+
+	if cliArgs.StartTime == "" {
+		err = ha.DeleteAllTaskCliArgs(s.etcdClient, taskName)
+		if err != nil {
+			return terror.Annotate(err, "while removing task command line arguments")
+		}
+	} else {
+		err = ha.PutTaskCliArgs(s.etcdClient, taskName, *req.SourceNameList, *cliArgs)
+		if err != nil {
+			return terror.Annotate(err, "while putting task command line arguments")
+		}
+	}
+	if release != nil {
 		release()
 	}
+
 	return s.scheduler.UpdateExpectSubTaskStage(pb.Stage_Running, taskName, *req.SourceNameList...)
 }
 

dm/dm/master/openapi_controller_test.go

@@ -22,6 +22,8 @@ import (
 	"context"
 	"testing"
 
+	"github.com/pingcap/tiflow/dm/pkg/ha"
+
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tiflow/dm/checker"
 	"github.com/pingcap/tiflow/dm/dm/config"
@@ -338,6 +340,24 @@ func (s *OpenAPIControllerSuite) TestTaskController() {
 		// stop success
 		s.Nil(server.stopTask(ctx, s.testTask.Name, openapi.StopTaskRequest{}))
 		s.Equal(server.scheduler.GetExpectSubTaskStage(s.testTask.Name, s.testSource.SourceName).Expect, pb.Stage_Stopped)
+
+		// start with cli args
+		startTime := "2022-05-05 12:12:12"
+		safeModeTimeDuration := "10s"
+		req = openapi.StartTaskRequest{
+			StartTime:            &startTime,
+			SafeModeTimeDuration: &safeModeTimeDuration,
+		}
+		s.Nil(server.startTask(ctx, s.testTask.Name, req))
+		taskCliConf, err := ha.GetTaskCliArgs(server.etcdClient, s.testTask.Name, s.testSource.SourceName)
+		s.Nil(err)
+		s.NotNil(taskCliConf)
+		s.Equal(startTime, taskCliConf.StartTime)
+		s.Equal(safeModeTimeDuration, taskCliConf.SafeModeDuration)
+
+		// stop success
+		s.Nil(server.stopTask(ctx, s.testTask.Name, openapi.StopTaskRequest{}))
+		s.Equal(server.scheduler.GetExpectSubTaskStage(s.testTask.Name, s.testSource.SourceName).Expect, pb.Stage_Stopped)
 	}
 
 	// delete

dm/openapi/spec/dm.yaml

@@ -2168,13 +2168,14 @@ components:
           description: whether to remove meta database in downstream database
         source_name_list:
           $ref: "#/components/schemas/SourceNameList"
-        # start_time:
-        #   type: string
-        #   example: "2006-01-02 15:04:05"
-        #   description: task start time
-        # safe_mode_time_duration:
-        #   example: "1s"
-        #   description: time duration of safe mode
+        start_time:
+          type: string
+          example: "2006-01-02 15:04:05"
+          description: task start time
+        safe_mode_time_duration:
+          type: string
+          example: "10s"
+          description: time duration of safe mode
     StopTaskRequest:
       type: object
       properties:

dm/syncer/safe_mode_test.go

@@ -62,10 +62,10 @@ func TestEnableSafeModeInitializationPhase(t *testing.T) {
 	require.True(t, s.safeMode.Enable())
 	s.Lock()
 	require.Nil(t, s.exitSafeModeTS) // not meet the first binlog
-	firstBinlogTS := int64(1)
+	firstBinlogTS := int64(1000)
 	require.NoError(t, s.initSafeModeExitTS(firstBinlogTS))
 	require.NotNil(t, s.exitSafeModeTS) // not meet the first binlog
-	require.Equal(t, int64(3), *s.exitSafeModeTS)
+	require.Equal(t, int64(3000), *s.exitSafeModeTS)
 	require.Equal(t, firstBinlogTS, *s.firstMeetBinlogTS)
 	s.Unlock()
 	require.NoError(t, s.checkAndExitSafeModeByBinlogTS(s.tctx, *s.exitSafeModeTS)) // not exit when binlog TS == exit TS

dm/syncer/syncer.go

@@ -2076,7 +2076,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		}
 
 		// set exitSafeModeTS when meet first binlog
-		if s.firstMeetBinlogTS == nil && s.cliArgs != nil && s.cliArgs.SafeModeDuration != "" {
+		if s.firstMeetBinlogTS == nil && s.cliArgs != nil && s.cliArgs.SafeModeDuration != "" && int64(e.Header.Timestamp) != 0 {
 			if checkErr := s.initSafeModeExitTS(int64(e.Header.Timestamp)); checkErr != nil {
 				return checkErr
 			}
@@ -2185,7 +2185,7 @@ func (s *Syncer) initSafeModeExitTS(firstBinlogTS int64) error {
 		return err
 	}
 	s.firstMeetBinlogTS = &firstBinlogTS
-	exitTS := firstBinlogTS + int64(duration.Seconds())
+	exitTS := firstBinlogTS + int64(duration.Seconds())*1000
 	s.exitSafeModeTS = &exitTS
 	s.tctx.L().Info("safe-mode will disable by task cli args", zap.Int64("ts", exitTS))
 	return nil

dm/tests/openapi/client/openapi_task_check

@@ -98,6 +98,66 @@ def create_noshard_task_success(task_name, tartget_table_name=""):
     print("create_noshard_task_success resp=", resp.json())
     assert resp.status_code == 201
 
+def create_incremental_task_with_gitd_success(task_name,binlog_name1,binlog_pos1,binlog_gtid1,binlog_name2,binlog_pos2,binlog_gtid2):
+    task = {
+        "name": task_name,
+        "task_mode": "incremental",
+        "meta_schema": "dm_meta",
+        "enhance_online_schema_change": True,
+        "on_duplicate": "error",
+        "target_config": {
+            "host": "127.0.0.1",
+            "port": 4000,
+            "user": "root",
+            "password": "",
+        },
+        "table_migrate_rule": [
+            {
+                "source": {
+                    "source_name": SOURCE1_NAME,
+                    "schema": "openapi",
+                    "table": "*",
+                },
+                "target": {"schema": "openapi", "table": ""},
+            },
+            {
+                "source": {
+                    "source_name": SOURCE2_NAME,
+                    "schema": "openapi",
+                    "table": "*",
+                },
+                "target": {"schema": "openapi", "table": ""},
+            },
+        ],
+         "source_config": {
+            "source_conf": [
+                {"source_name": SOURCE1_NAME},
+                {"source_name": SOURCE2_NAME},
+            ],
+         },
+    }
+
+    if binlog_pos1 != "":
+        task["source_config"] = {
+            "source_conf": [
+                {
+                    "source_name": SOURCE1_NAME,
+                    "binlog_name": binlog_name1,
+                    "binlog_pos": int(binlog_pos1),
+                    "binlog_gtid": binlog_gtid1,
+                },
+                {
+                    "source_name": SOURCE2_NAME,
+                    "binlog_name": binlog_name2,
+                    "binlog_pos": int(binlog_pos2),
+                    "binlog_gtid": binlog_gtid2,
+                },
+            ],
+        }
+
+    resp = requests.post(url=API_ENDPOINT, json={"task": task})
+    print("create_incremental_task_with_gitd_success resp=", resp.json())
+    assert resp.status_code == 201
 
 def create_shard_task_success():
     task = {
@@ -170,6 +230,21 @@ def start_task_success(task_name, source_name):
         print("start_task_failed resp=", resp.json())
     assert resp.status_code == 200
 
+def start_task_with_condition(task_name, start_time, duration, is_success, check_result):
+    url = API_ENDPOINT + "/" + task_name + "/start"
+    req = {"start_time": start_time, "safe_mode_time_duration": duration}
+    resp = requests.post(url=url, json=req)
+    if is_success == "success":
+        if resp.status_code != 200:
+            print("start_task_with_condition_failed resp=", resp.json())
+        assert resp.status_code == 200
+        print("start_task_with_condition success")
+    else:
+        if resp.status_code == 200:
+             print("start_task_with_condition_failed resp should not be 200")
+        assert resp.status_code == 400
+        print("start_task_with_condition resp=", resp.json())
+        assert check_result in resp.json()["error_msg"]
 
 def stop_task_success(task_name, source_name):
     url = API_ENDPOINT + "/" + task_name + "/stop"
@@ -568,10 +643,12 @@ if __name__ == "__main__":
         "create_task_failed": create_task_failed,
         "create_noshard_task_success": create_noshard_task_success,
         "create_shard_task_success": create_shard_task_success,
+        "create_incremental_task_with_gitd_success": create_incremental_task_with_gitd_success,
         "delete_task_failed": delete_task_failed,
         "delete_task_success": delete_task_success,
         "delete_task_with_force_success": delete_task_with_force_success,
         "start_task_success": start_task_success,
+        "start_task_with_condition": start_task_with_condition,
         "stop_task_success": stop_task_success,
         "get_task_list": get_task_list,
         "get_task_list_with_status": get_task_list_with_status,