dmctl: use dmctl binary directly in integration test

Makefile

@@ -99,6 +99,10 @@ dm_integration_test_build:
 		-coverpkg=github.com/pingcap/dm/... \
 		-o bin/dm-master.test github.com/pingcap/dm/cmd/dm-master \
 		|| { $(FAILPOINT_DISABLE); exit 1; }
+	$(GOTEST) -c -race -cover -covermode=atomic \
+		-coverpkg=github.com/pingcap/dm/... \
+		-o bin/dmctl.test github.com/pingcap/dm/cmd/dm-ctl \
+		|| { $(FAILPOINT_DISABLE); exit 1; }
 	$(GOTEST) -c -race -cover -covermode=atomic \
 		-coverpkg=github.com/pingcap/dm/... \
 		-o bin/dm-tracer.test github.com/pingcap/dm/cmd/dm-tracer \

tests/dmctl_tools/dmctl_start_task.go → cmd/dm-ctl/main_test.go

@@ -13,21 +13,25 @@
 
 package main
 
+// Reference: https://dzone.com/articles/measuring-integration-test-coverage-rate-in-pouchc
+
 import (
-	"context"
 	"os"
-
-	"github.com/pingcap/dm/tests/utils"
+	"strings"
+	"testing"
 )
 
-func main() {
-	cli, err := utils.CreateDmCtl("127.0.0.1:8261")
-	if err != nil {
-		utils.ExitWithError(err)
-	}
-	conf := os.Args[1]
-	err = utils.StartTask(context.Background(), cli, conf, nil)
-	if err != nil {
-		utils.ExitWithError(err)
+func TestRunMain(t *testing.T) {
+	var args []string
+	for _, arg := range os.Args {
+		switch {
+		case arg == "DEVEL":
+		case strings.HasPrefix(arg, "-test."):
+		default:
+			args = append(args, arg)
+		}
 	}
+
+	os.Args = args
+	main()
 }

dm/ctl/master/operate_relay.go → dm/ctl/common/operate_relay.go

@@ -11,20 +11,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package master
+package common
 
 import (
 	"context"
 
-	"github.com/pingcap/dm/dm/ctl/common"
 	"github.com/pingcap/dm/dm/pb"
 )
 
-// operateRelay does operation on relay unit
-func operateRelay(op pb.RelayOp, workers []string) (*pb.OperateWorkerRelayResponse, error) {
+// OperateRelay does operation on relay unit
+func OperateRelay(op pb.RelayOp, workers []string) (*pb.OperateWorkerRelayResponse, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	cli := common.MasterClient()
+	cli := MasterClient()
 	return cli.OperateWorkerRelayTask(ctx, &pb.OperateWorkerRelayRequest{
 		Op:      op,
 		Workers: workers,

dm/ctl/master/operate_task.go → dm/ctl/common/operate_task.go

@@ -11,20 +11,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package master
+package common
 
 import (
 	"context"
 
-	"github.com/pingcap/dm/dm/ctl/common"
 	"github.com/pingcap/dm/dm/pb"
 )
 
-// operateTask does operation on task
-func operateTask(op pb.TaskOp, name string, workers []string) (*pb.OperateTaskResponse, error) {
+// OperateTask does operation on task
+func OperateTask(op pb.TaskOp, name string, workers []string) (*pb.OperateTaskResponse, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	cli := common.MasterClient()
+	cli := MasterClient()
 	return cli.OperateTask(ctx, &pb.OperateTaskRequest{
 		Op:      op,
 		Name:    name,

dm/ctl/ctl.go

@@ -67,10 +67,10 @@ func Start(args []string) {
 		master.NewSwitchRelayMasterCmd(),
 		master.NewPauseRelayCmd(),
 		master.NewResumeRelayCmd(),
-		//master.NewStopRelayCmd(),
 		master.NewUpdateMasterConfigCmd(),
 		master.NewUpdateRelayCmd(),
 		master.NewPurgeRelayCmd(),
+		master.NewMigrateRelayCmd(),
 	)
 
 	rootCmd.SetArgs(args)

dm/ctl/master/pause_relay.go

@@ -16,10 +16,11 @@ package master
 import (
 	"fmt"
 
-	"github.com/pingcap/dm/dm/ctl/common"
-	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/errors"
 	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
 )
 
 // NewPauseRelayCmd creates a PauseRelay command
@@ -49,7 +50,7 @@ func pauseRelayFunc(cmd *cobra.Command, _ []string) {
 		return
 	}
 
-	resp, err := operateRelay(pb.RelayOp_PauseRelay, workers)
+	resp, err := common.OperateRelay(pb.RelayOp_PauseRelay, workers)
 	if err != nil {
 		common.PrintLines("can not pause relay unit:\n%v", errors.ErrorStack(err))
 		return

dm/ctl/master/pause_task.go

@@ -16,10 +16,11 @@ package master
 import (
 	"fmt"
 
-	"github.com/pingcap/dm/dm/ctl/common"
-	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/errors"
 	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
 )
 
 // NewPauseTaskCmd creates a PauseTask command
@@ -46,7 +47,7 @@ func pauseTaskFunc(cmd *cobra.Command, _ []string) {
 		return
 	}
 
-	resp, err := operateTask(pb.TaskOp_Pause, name, workers)
+	resp, err := common.OperateTask(pb.TaskOp_Pause, name, workers)
 	if err != nil {
 		common.PrintLines("can not pause task %s:\n%v", name, errors.ErrorStack(err))
 		return

dm/ctl/master/resume_relay.go

@@ -16,10 +16,11 @@ package master
 import (
 	"fmt"
 
-	"github.com/pingcap/dm/dm/ctl/common"
-	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/errors"
 	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
 )
 
 // NewResumeRelayCmd creates a ResumeRelay command
@@ -49,7 +50,7 @@ func resumeRelayFunc(cmd *cobra.Command, _ []string) {
 		return
 	}
 
-	resp, err := operateRelay(pb.RelayOp_ResumeRelay, workers)
+	resp, err := common.OperateRelay(pb.RelayOp_ResumeRelay, workers)
 	if err != nil {
 		common.PrintLines("can not resume relay unit:\n%v", errors.ErrorStack(err))
 		return

dm/ctl/master/resume_task.go

@@ -16,10 +16,11 @@ package master
 import (
 	"fmt"
 
-	"github.com/pingcap/dm/dm/ctl/common"
-	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/errors"
 	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
 )
 
 // NewResumeTaskCmd creates a ResumeTask command
@@ -46,7 +47,7 @@ func resumeTaskFunc(cmd *cobra.Command, _ []string) {
 		return
 	}
 
-	resp, err := operateTask(pb.TaskOp_Resume, name, workers)
+	resp, err := common.OperateTask(pb.TaskOp_Resume, name, workers)
 	if err != nil {
 		common.PrintLines("can not resume task %s:\n%v", name, errors.ErrorStack(err))
 		return

dm/ctl/master/sql_skip.go

@@ -18,10 +18,11 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/pingcap/dm/dm/ctl/common"
-	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/errors"
 	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
 )
 
 // NewSQLSkipCmd creates a SQLSkip command

dm/ctl/master/stop_task.go

@@ -16,10 +16,11 @@ package master
 import (
 	"fmt"
 
-	"github.com/pingcap/dm/dm/ctl/common"
-	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/errors"
 	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
 )
 
 // NewStopTaskCmd creates a StopTask command
@@ -46,7 +47,7 @@ func stopTaskFunc(cmd *cobra.Command, _ []string) {
 		return
 	}
 
-	resp, err := operateTask(pb.TaskOp_Stop, name, workers)
+	resp, err := common.OperateTask(pb.TaskOp_Stop, name, workers)
 	if err != nil {
 		common.PrintLines("can not stop task %s:\n%v", name, errors.ErrorStack(err))
 		return

tests/README.md

@@ -51,6 +51,7 @@ Several convenient commands are provided:
 
 * `run_dm_master <WORKDIR> <PORT> <CONFIG>` — Starts `dm-master` using config provided, on given port, running in workdir.
 * `run_dm_worker <WORKDIR> <PORT> <CONFIG>` — Starts `dm-worker` using config provided, on given port, running in workdir.
+* `run_dm_ctl <WORKDIR> <MASTER_ADDR> <DMCTL_COMMAND>` - Runs `dmctl` with given command in non-interactive mode.
 * `run_sql <SQL> <PORT>` — Executes an SQL query in database based on port provided
 * `run_sql_file <path_to_SQL_file> <HOST> <PORT>` — Executes all SQLs in given file to the database on port provided
 * `run_sql_file_online_ddl <path_to_SQL_file> <HOST> <PORT> <DB> <ONLINE DDL TOOL>` — Executes all SQLs in given file, will auto switch DDL to online DDL command.

tests/_utils/check_sync_diff

@@ -1,10 +1,15 @@
 #!/bin/bash
 # parameter 1: work directory
 # parameter 2: config file for sync_diff_inspector
+# parameter 3: max check times
 
 workdir=$1
 conf=$2
-check_time=10
+if [ $# -ge 3 ]; then
+    check_time=$3
+else
+    check_time=10
+fi
 
 PWD=$(pwd)
 binary=$PWD/bin/sync_diff_inspector

tests/_utils/run_dm_ctl

@@ -0,0 +1,31 @@
+#!/bin/bash
+# tools to run dmctl from command line
+# parameter 1: work directory
+# parameter 2: master-addr port
+# parameter 3: command
+# parameter 4...: check output content and count
+
+workdir=$1
+master_addr=$2
+cmd=$3
+
+shift 3
+
+PWD=$(pwd)
+binary=$PWD/bin/dmctl.test
+ts=$(date +"%s")
+dmctl_log=$workdir/dmctl.$ts.log
+echo "dmctl test cmd: \"$cmd\""
+echo "$cmd" | $binary -test.coverprofile="$TEST_DIR/cov.$TEST_NAME.dmctl.$ts.out" DEVEL -master-addr=$master_addr > $dmctl_log 2>&1
+
+for ((i=1; i<$#; i+=2)); do
+    j=$((i+1))
+    value=${!i}
+    expected=${!j}
+    got=$(sed "s/$value/$value\n/g" $dmctl_log | grep -c "$value")
+    if [ "$got" != "$expected" ]; then
+        echo "command: $cmd $value count: $got != expected: $expected"
+        cat $dmctl_log
+        exit 1
+    fi
+done

tests/_utils/test_prepare

@@ -33,3 +33,40 @@ fi
 function join_string() {
     local IFS="$1"; shift; echo "$*";
 }
+
+# shortcut for start task on one DM-worker
+function dmctl_start_task_standalone() {
+    if [ $# -ge 1 ]; then
+        task_conf=$1
+    else
+        task_conf="$cur/conf/dm-task.yaml"
+    fi
+    run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+        "start-task $task_conf" \
+        "\"result\": true" 2 \
+        "\"worker\": \"127.0.0.1:$WORKER1_PORT\"" 1
+}
+
+# shortcut for start task on two DM-workers
+function dmctl_start_task() {
+    if [ $# -ge 1 ]; then
+        task_conf=$1
+    else
+        task_conf="$cur/conf/dm-task.yaml"
+    fi
+    run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+        "start-task $task_conf" \
+        "\"result\": true" 3 \
+        "\"worker\": \"127.0.0.1:$WORKER1_PORT\"" 1 \
+        "\"worker\": \"127.0.0.1:$WORKER2_PORT\"" 1
+}
+
+# shortcut for stop task on two DM-workers
+function dmctl_stop_task() {
+    task_name=$1
+    run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+        "stop-task $task_name" \
+        "\"result\": true" 3 \
+        "\"worker\": \"127.0.0.1:$WORKER1_PORT\"" 1 \
+        "\"worker\": \"127.0.0.1:$WORKER2_PORT\"" 1
+}