*: show an unsynced message for workers which didn't get any shard DDL

dm/master/server.go

@@ -691,6 +691,45 @@ func (s *Server) QueryStatus(ctx context.Context, req *pb.QueryStatusListRequest
 
 	resps := s.getStatusFromWorkers(ctx, sources, req.Name, queryRelayWorker)
 
+	// adjust unsynced field in sync status, task-name -> set(target-table)
+	// because if a DM-worker doesn't receive any shard DDL, it doesn't even know it's unsynced for itself
+	unsyncedMap := make(map[string]map[string]struct{})
+	for _, resp := range resps {
+		for _, subtaskStatus := range resp.SubTaskStatus {
+			syncStatus := subtaskStatus.GetSync()
+			if syncStatus == nil {
+				continue
+			}
+			for _, group := range syncStatus.UnresolvedGroups {
+				if _, ok := unsyncedMap[subtaskStatus.Name]; !ok {
+					unsyncedMap[subtaskStatus.Name] = make(map[string]struct{})
+				}
+				unsyncedMap[subtaskStatus.Name][group.Target] = struct{}{}
+			}
+		}
+	}
+	for _, resp := range resps {
+		for _, subtaskStatus := range resp.SubTaskStatus {
+			syncStatus := subtaskStatus.GetSync()
+			if syncStatus == nil {
+				continue
+			}
+			found := make(map[string]struct{}, len(syncStatus.UnresolvedGroups))
+			for _, group := range syncStatus.UnresolvedGroups {
+				found[group.Target] = struct{}{}
+			}
+			for target := range unsyncedMap[subtaskStatus.Name] {
+				if _, ok := found[target]; ok {
+					continue
+				}
+				syncStatus.UnresolvedGroups = append(syncStatus.UnresolvedGroups, &pb.ShardingGroup{
+					Target:   target,
+					Unsynced: []string{"this DM-worker doesn't receive any shard DDL of this group"},
+				})
+			}
+		}
+	}
+
 	workerRespMap := make(map[string][]*pb.QueryStatusResponse, len(sources))
 	for _, workerResp := range resps {
 		workerRespMap[workerResp.SourceStatus.Source] = append(workerRespMap[workerResp.SourceStatus.Source], workerResp)

dm/worker/worker.go

@@ -506,8 +506,10 @@ func (w *Worker) QueryStatus(ctx context.Context, name string) ([]*pb.SubTaskSta
 		w.dbMutex.Lock()
 		if w.db == nil {
 			var err error
-			w.db, err = conn.DefaultDBProvider.Apply(w.cfg.From)
+			w.l.Info("will open a connection to get master status", zap.Any("upstream config", w.cfg.From))
+			w.db, err = conn.DefaultDBProvider.Apply(w.cfg.DecryptPassword().From)
 			if err != nil {
+				w.l.Error("can't open a connection to get master status", zap.Error(err))
 				w.dbMutex.Unlock()
 				return subtaskStatus, relayStatus, err
 			}

tests/sequence_sharding_removemeta/run.sh

@@ -44,6 +44,10 @@ function run() {
         "show-ddl-locks" \
         "\"ID\": \"$lock_id\"" 1 \
         "$ddl" 1
+    run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+        "query-status $TEST_NAME" \
+        "this DM-worker doesn't receive any shard DDL of this group" 0 \
+        "\"masterBinlog\": \"\"" 0
     check_metric $MASTER_PORT 'dm_master_ddl_state_number{task="sequence_sharding_removemeta",type="Un-synced"}' 3 0 2
 
     dmctl_stop_task $task_name
@@ -70,6 +74,9 @@ function run() {
         "show-ddl-locks" \
         "\"ID\": \"$lock_id\"" 1 \
         "$ddl" 1
+    run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+        "query-status $TEST_NAME" \
+        "this DM-worker doesn't receive any shard DDL of this group" 1
 
     check_metric $MASTER_PORT 'dm_master_ddl_state_number{task="sequence_sharding_removemeta",type="Un-synced"}' 3 0 2
     dmctl_stop_task $task_name