openapi(dm): add dump status openapi.go (#4033)

dm/dm/common/common.go

@@ -96,7 +96,6 @@ var (
 	// task config template is used to generate task config before user create a real task by openapi. user can modify eg:
 	// import from running tasks/create/update/delete the template and those changes will not affect the running tasks.
 	OpenAPITaskTemplateKeyAdapter KeyAdapter = keyHexEncoderDecoder("/dm-master/openapi-task-template/")
-
 	// TaskCliArgsKeyAdapter is used to store the command line arguments of task. They are different from the task
 	// config because the command line arguments may be expected to take effect only once when failover.
 	// kv: Encode(task-name, source-id) -> TaskCliArgs.

dm/dm/master/openapi.go

@@ -985,6 +985,16 @@ func getOpenAPISubtaskStatusByTaskName(taskName string,
 				}
 			}
 		}
+		// add dump status
+		if dumpS := subTaskStatus.GetDump(); dumpS != nil {
+			openapiSubTaskStatus.DumpStatus = &openapi.DumpStatus{
+				CompletedTables:   dumpS.CompletedTables,
+				EstimateTotalRows: dumpS.EstimateTotalRows,
+				FinishedBytes:     dumpS.FinishedBytes,
+				FinishedRows:      dumpS.FinishedRows,
+				TotalTables:       dumpS.TotalTables,
+			}
+		}
 		subTaskStatusList = append(subTaskStatusList, openapiSubTaskStatus)
 	}
 	return subTaskStatusList, nil

dm/dm/master/openapi_test.go

@@ -548,7 +548,9 @@ func (t *openAPISuite) TestTaskAPI(c *check.C) {
 	c.Assert(resultTaskStatus.Data[0].Name, check.Equals, task.Name)
 	c.Assert(resultTaskStatus.Data[0].Stage, check.Equals, pb.Stage_Running.String())
 	c.Assert(resultTaskStatus.Data[0].WorkerName, check.Equals, workerName1)
-
+	c.Assert(resultTaskStatus.Data[0].DumpStatus.CompletedTables, check.Equals, float64(0))
+	c.Assert(resultTaskStatus.Data[0].DumpStatus.TotalTables, check.Equals, int64(1))
+	c.Assert(resultTaskStatus.Data[0].DumpStatus.EstimateTotalRows, check.Equals, float64(10))
 	// get task status with source name
 	taskStatusURL = fmt.Sprintf("%s/%s/status?source_name_list=%s", taskURL, task.Name, source1Name)
 	result = testutil.NewRequest().Get(taskStatusURL).GoWithHTTPHandler(t.testT, s.openapiHandles)
@@ -857,6 +859,19 @@ func mockTaskQueryStatus(
 					},
 				},
 			},
+			{
+				Stage: pb.Stage_Running,
+				Name:  taskName,
+				Status: &pb.SubTaskStatus_Dump{
+					Dump: &pb.DumpStatus{
+						CompletedTables:   0.0,
+						EstimateTotalRows: 10.0,
+						FinishedBytes:     0.0,
+						FinishedRows:      5.0,
+						TotalTables:       1,
+					},
+				},
+			},
 		},
 	}
 	mockWorkerClient.EXPECT().QueryStatus(

dm/openapi/spec/dm.yaml

@@ -1232,6 +1232,31 @@ components:
         - "synced"
         - "binlog_type"
         - "seconds_behind_master"
+    DumpStatus:
+      type: object
+      description: "status of dump unit"
+      properties:
+        total_tables:
+          type: integer
+          format: int64
+        completed_tables:
+          type: number
+          format: double
+        finished_bytes:
+          type: number
+          format: double
+        finished_rows:
+          type: number
+          format: double
+        estimate_total_rows:
+          type: number
+          format: double
+      required:
+        - "total_tables"
+        - "completed_tables"
+        - "finished_bytes"
+        - "finished_rows"
+        - "estimate_total_rows" 
     SubTaskStatus:
       type: object
       properties:
@@ -1260,6 +1285,9 @@ components:
         sync_status:
           nullable: true
           $ref: "#/components/schemas/SyncStatus"
+        dump_status:
+          nullable: true
+          $ref: "#/components/schemas/DumpStatus"
       required:
         - "name"
         - "source_name"

dm/tests/openapi/client/openapi_task_check

@@ -295,6 +295,15 @@ def stop_task_success(task_name):
     print("stop_task_success")
 
 
+def check_noshard_task_dump_status_success(task_name, total):
+    url = API_ENDPOINT + "/" + task_name + "/status"
+    resp = requests.get(url=url)
+    data = resp.json()
+    assert resp.status_code == 200
+    print("check_dump_status_success resp=", data)
+    assert data["data"][0]["dump_status"]["finished_bytes"] == int(total)
+
+
 if __name__ == "__main__":
     FUNC_MAP = {
         "start_task_failed": start_task_failed,
@@ -310,6 +319,7 @@ if __name__ == "__main__":
         "get_illegal_char_task_status_failed": get_illegal_char_task_status_failed,
         "get_task_status_success": get_task_status_success,
         "operate_schema_and_table_success": operate_schema_and_table_success,
+        "check_noshard_task_dump_status_success": check_noshard_task_dump_status_success,
     }
 
     func = FUNC_MAP[sys.argv[1]]