*: support sourceID on operate-source stop

dm/ctl/master/operate_source.go

@@ -31,7 +31,11 @@ import (
 // NewOperateSourceCmd creates a OperateSource command
 func NewOperateSourceCmd() *cobra.Command {
 	cmd := &cobra.Command{
-		Use:   "operate-source <operate-type> [config-file ...] [--print-sample-config]",
+		// the leading space could provide more friendly terminal printing
+		// cobra.Command doesn't support multi-line usage well
+		Use: `operate-source create|update [config-file ...] [--print-sample-config] [flags]
+        operate-source stop [config-file | sourceID ...] [--print-sample-config] [flags]
+        operate-source show [--print-sample-config] [flags]`,
 		Short: "create/update/stop/show upstream MySQL/MariaDB source",
 		RunE:  operateSourceFunc,
 	}
@@ -96,24 +100,30 @@ func operateSourceFunc(cmd *cobra.Command, _ []string) (err error) {
 		return
 	}
 
-	contents := make([]string, len(cmd.Flags().Args())-1)
+	contents := make([]string, 0, len(cmd.Flags().Args())-1)
+	sourceID := make([]string, 0, len(cmd.Flags().Args())-1)
 	for i := 1; i < len(cmd.Flags().Args()); i++ {
-		configFile := cmd.Flags().Arg(i)
+		arg := cmd.Flags().Arg(i)
 		var content []byte
-		content, err = common.GetFileContent(configFile)
+		content, err = common.GetFileContent(arg)
 		if err != nil {
+			if op == pb.SourceOp_StopSource {
+				sourceID = append(sourceID, arg)
+				continue
+			}
 			return
 		}
-		contents[i-1] = string(content)
+		contents = append(contents, string(content))
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	cli := common.MasterClient()
 	resp, err := cli.OperateSource(ctx, &pb.OperateSourceRequest{
-		Config: contents,
-		Op:     op,
+		Config:   contents,
+		Op:       op,
+		SourceID: sourceID,
 	})
 	if err != nil {
 		return

dm/master/server.go

@@ -1469,9 +1469,17 @@ func (s *Server) OperateSource(ctx context.Context, req *pb.OperateSourceRequest
 		resp.Msg = "Update worker config is not supported by dm-ha now"
 		return resp, nil
 	case pb.SourceOp_StopSource:
+		toRemove := make([]string, 0, len(cfgs)+len(req.SourceID))
+		for _, sid := range req.SourceID {
+			toRemove = append(toRemove, sid)
+		}
 		for _, cfg := range cfgs {
-			boundM[cfg.SourceID] = s.scheduler.GetWorkerBySource(cfg.SourceID)
-			err := s.scheduler.RemoveSourceCfg(cfg.SourceID)
+			toRemove = append(toRemove, cfg.SourceID)
+		}
+
+		for _, sid := range toRemove {
+			boundM[sid] = s.scheduler.GetWorkerBySource(sid)
+			err := s.scheduler.RemoveSourceCfg(sid)
 			// TODO(lance6716):
 			// user could not copy-paste same command if encounter error halfway:
 			// `operate-source stop  correct-id-1     wrong-id-2`