dm: support dump task mode in openapi (#11729)

close #9230

dm/openapi/spec/dm.yaml

@@ -1825,6 +1825,7 @@ components:
             - "full"
             - "incremental"
             - "all"
+            - "dump"
         shard_mode:
           type: string
           description: the way to coordinate DDL

dm/tests/_utils/run_dm_ctl_with_retry

@@ -8,8 +8,15 @@
 workdir=$1
 master_addr=$2
 cmd=$3
+retry_times=10 # Default retry times
 
-shift 3
+# Validate retry_times is a number
+if [[ $4 =~ ^[0-9]+$ ]] && [[ $# -gt 4 ]]; then
+	retry_times=$4
+	shift 4
+else
+	shift 3
+fi
 
 PWD=$(pwd)
 binary=$PWD/bin/dmctl.test
@@ -18,8 +25,8 @@ dmctl_log=$workdir/dmctl.$ts.log
 pid=$$
 
 all_matched=true
-echo "dmctl test cmd: \"$cmd\""
-for ((k = 0; k < 10; k++)); do
+echo "dmctl test cmd: \"$cmd\" with max $retry_times retries"
+for ((k = 0; k < retry_times; k++)); do
 	echo "$cmd" | $binary -test.coverprofile="$TEST_DIR/cov.$TEST_NAME.dmctl.$ts.$pid.out" DEVEL --master-addr=$master_addr >$dmctl_log 2>&1
 	all_matched=true
 	for ((i = 1; i < $#; i += 2)); do

dm/tests/openapi/client/openapi_task_check

@@ -268,6 +268,45 @@ def create_shard_task_success():
     print("create_shard_task_success resp=", resp.json())
     assert resp.status_code == 201
 
+def create_dump_task_success():
+    task = {
+        "name": "test-dump",
+        "task_mode": "dump",
+        "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": "t"},
+            }
+        ],
+        "source_config": {
+            "full_migrate_conf": {
+                "export_threads": 4,
+                "import_threads": 16,
+                "data_dir": "./exported_data",
+                "consistency": "auto",
+            },
+            "incr_migrate_conf": {"repl_threads": 16, "repl_batch": 100},
+            "source_conf": [
+                {"source_name": SOURCE1_NAME},
+            ],
+        }
+    }
+    resp = requests.post(url=API_ENDPOINT, json={"task": task})
+    print("create_dump_task_success resp=", resp.json())
+    assert resp.status_code == 201
 
 def start_task_success(task_name, source_name):
     url = API_ENDPOINT + "/" + task_name + "/start"
@@ -703,6 +742,7 @@ 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_dump_task_success": create_dump_task_success,
         "create_incremental_task_with_gtid_success": create_incremental_task_with_gtid_success,
         "delete_task_failed": delete_task_failed,
         "delete_task_success": delete_task_success,

dm/tests/openapi/run.sh

@@ -24,6 +24,12 @@ function init_noshard_data() {
 	run_sql_source2 "INSERT INTO openapi.t2(i,j) VALUES (3, 4);"
 }
 
+function init_dump_data() {
+
+	run_sql_source1 "CREATE TABLE openapi.t1(i TINYINT, j INT UNIQUE KEY);"
+	run_sql_source1 "INSERT INTO openapi.t1(i,j) VALUES (1, 2),(3,4);"
+}
+
 function init_shard_data() {
 	run_sql_source1 "CREATE TABLE openapi.t(i TINYINT, j INT UNIQUE KEY);"
 	run_sql_source2 "CREATE TABLE openapi.t(i TINYINT, j INT UNIQUE KEY);"
@@ -176,6 +182,52 @@ function test_relay() {
 
 }
 
+function test_dump_task() {
+	echo ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>START TEST OPENAPI: dump TASK"
+	prepare_database
+
+	task_name="test-dump"
+
+	# create source successfully
+	openapi_source_check "create_source1_success"
+	# get source list success
+	openapi_source_check "list_source_success" 1
+
+	# create source successfully
+	openapi_source_check "create_source2_success"
+	# get source list success
+	openapi_source_check "list_source_success" 2
+
+	# get source status success
+	openapi_source_check "get_source_status_success" "mysql-01"
+
+	# create task success: not valid task create request
+	openapi_task_check "create_task_failed"
+
+	# create dump task success
+	openapi_task_check "create_dump_task_success"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status $task_name" \
+		"\"stage\": \"Stopped\"" 1
+
+	init_dump_data
+
+	# start dump task success
+	openapi_task_check "start_task_success" $task_name ""
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status $task_name" \
+		"\"stage\": \"Running\"" 1
+
+	# wait dump task finish
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status $task_name" 100 \
+		"\"stage\": \"Finished\"" 1
+
+	clean_cluster_sources_and_tasks
+	echo ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>TEST OPENAPI: dump TASK"
+
+}
+
 function test_shard_task() {
 	echo ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>START TEST OPENAPI: SHARD TASK"
 	prepare_database
@@ -1040,6 +1092,7 @@ function run() {
 	test_shard_task
 	test_multi_tasks
 	test_noshard_task
+	test_dump_task
 	test_task_templates
 	test_noshard_task_dump_status
 	test_complex_operations_of_source_and_task