Improve error logs

Signed-off-by: Matt Lord <[email protected]>

Author: mattlord

go/vt/vttablet/tabletmanager/vdiff/table_differ.go

@@ -122,18 +122,18 @@ func (td *tableDiffer) initialize(ctx context.Context) error {
 
 	targetKeyspace := td.wd.ct.vde.thisTablet.Keyspace
 	lockName := fmt.Sprintf("%s/%s", targetKeyspace, td.wd.ct.workflow)
-	log.Infof("Locking workflow %s", lockName)
+	log.Infof("Locking workflow %s for vdiff %s", lockName, td.wd.ct.uuid)
 	ctx, unlock, lockErr := td.wd.ct.ts.LockName(ctx, lockName, "vdiff")
 	if lockErr != nil {
-		log.Errorf("Locking workfkow %s failed: %v", lockName, lockErr)
+		log.Errorf("Locking workfkow %s for vdiff %s failed: %v", lockName, td.wd.ct.uuid, lockErr)
 		return lockErr
 	}
 
 	var err error
 	defer func() {
 		unlock(&err)
 		if err != nil {
-			log.Errorf("Unlocking workflow %s failed: %v", lockName, err)
+			log.Errorf("Unlocking workflow %s for vdiff %s failed: %v", lockName, td.wd.ct.uuid, err)
 		}
 	}()
 
@@ -143,12 +143,12 @@ func (td *tableDiffer) initialize(ctx context.Context) error {
 	defer func() {
 		// We use a new context as we want to reset the state even
 		// when the parent context has timed out or been canceled.
-		log.Infof("Restarting the %q VReplication workflow on target tablets in keyspace %q",
-			td.wd.ct.workflow, targetKeyspace)
+		log.Infof("Restarting the %q VReplication workflow for vdiff %s on target tablets in keyspace %q",
+			td.wd.ct.workflow, td.wd.ct.uuid, targetKeyspace)
 		restartCtx, restartCancel := context.WithTimeout(context.Background(), BackgroundOperationTimeout)
 		defer restartCancel()
 		if err := td.restartTargetVReplicationStreams(restartCtx); err != nil {
-			log.Errorf("error restarting target streams: %v", err)
+			log.Errorf("error restarting target streams for vdiff %s: %v", td.wd.ct.uuid, err)
 		}
 	}()
 
@@ -175,7 +175,7 @@ func (td *tableDiffer) initialize(ctx context.Context) error {
 }
 
 func (td *tableDiffer) stopTargetVReplicationStreams(ctx context.Context, dbClient binlogplayer.DBClient) error {
-	log.Infof("stopTargetVReplicationStreams")
+	log.Infof("stopTargetVReplicationStreams for vdiff %s", td.wd.ct.uuid)
 	ct := td.wd.ct
 	query := "update _vt.vreplication set state = 'Stopped', message='for vdiff' " + ct.workflowFilter
 	if _, err := ct.vde.vre.Exec(query); err != nil {
@@ -332,7 +332,7 @@ func (td *tableDiffer) syncTargetStreams(ctx context.Context) error {
 			return err
 		}
 		if err := ct.vde.vre.WaitForPos(waitCtx, source.vrID, source.snapshotPosition); err != nil {
-			log.Errorf("WaitForPosition error: %d: %s", source.vrID, err)
+			log.Errorf("WaitForPosition for vdiff %s error: %d: %s", td.wd.ct.uuid, source.vrID, err)
 			return vterrors.Wrapf(err, "WaitForPosition for stream id %d", source.vrID)
 		}
 		return nil
@@ -350,7 +350,8 @@ func (td *tableDiffer) startTargetDataStream(ctx context.Context) error {
 	go td.streamOneShard(ctx, ct.targetShardStreamer, td.tablePlan.targetQuery, td.lastTargetPK, gtidch)
 	gtid, ok := <-gtidch
 	if !ok {
-		log.Infof("streaming error: %v", ct.targetShardStreamer.err)
+		log.Errorf("VDiff %s streaming error on target tablet %s: %v",
+			td.wd.ct.uuid, topoproto.TabletAliasString(ct.targetShardStreamer.tablet.Alias), ct.targetShardStreamer.err)
 		return ct.targetShardStreamer.err
 	}
 	ct.targetShardStreamer.snapshotPosition = gtid
@@ -366,6 +367,8 @@ func (td *tableDiffer) startSourceDataStreams(ctx context.Context) error {
 
 		gtid, ok := <-gtidch
 		if !ok {
+			log.Errorf("VDiff %s streaming error on source tablet %s: %v",
+				td.wd.ct.uuid, topoproto.TabletAliasString(source.tablet.Alias), source.err)
 			return source.err
 		}
 		source.snapshotPosition = gtid
@@ -381,7 +384,7 @@ func (td *tableDiffer) restartTargetVReplicationStreams(ctx context.Context) err
 	ct := td.wd.ct
 	query := fmt.Sprintf("update _vt.vreplication set state='Running', message='', stop_pos='' where db_name=%s and workflow=%s",
 		encodeString(ct.vde.dbName), encodeString(ct.workflow))
-	log.Infof("Restarting the %q VReplication workflow using %q", ct.workflow, query)
+	log.Infof("Restarting the %q VReplication workflow for vdiff %s using %q", ct.workflow, td.wd.ct.uuid, query)
 	var err error
 	// Let's retry a few times if we get a retryable error.
 	for i := 1; i <= 3; i++ {
@@ -396,11 +399,11 @@ func (td *tableDiffer) restartTargetVReplicationStreams(ctx context.Context) err
 }
 
 func (td *tableDiffer) streamOneShard(ctx context.Context, participant *shardStreamer, query string, lastPK *querypb.QueryResult, gtidch chan string) {
-	log.Infof("streamOneShard Start on %s using query: %s", participant.tablet.Alias.String(), query)
+	log.Infof("streamOneShard Start for vdiff %s on %s using query: %s", td.wd.ct.uuid, participant.tablet.Alias.String(), query)
 	td.wgShardStreamers.Add(1)
 
 	defer func() {
-		log.Infof("streamOneShard End on %s (err: %v)", participant.tablet.Alias.String(), participant.err)
+		log.Infof("streamOneShard for vdiff %s End on %s (err: %v)", td.wd.ct.uuid, participant.tablet.Alias.String(), participant.err)
 		select {
 		case <-ctx.Done():
 		default:
@@ -542,7 +545,7 @@ func (td *tableDiffer) diff(ctx context.Context, coreOpts *tabletmanagerdatapb.V
 	// Save our progress when we finish the run.
 	defer func() {
 		if err := td.updateTableProgress(dbClient, dr, lastProcessedRow); err != nil {
-			log.Errorf("Failed to update vdiff progress on %s table: %v", td.table.Name, err)
+			log.Errorf("Failed to update vdiff %s progress on %s table: %v", td.wd.ct.uuid, td.table.Name, err)
 		}
 		globalStats.RowsDiffedCount.Add(dr.ProcessedRows)
 	}()
@@ -579,15 +582,15 @@ func (td *tableDiffer) diff(ctx context.Context, coreOpts *tabletmanagerdatapb.V
 
 		if !mismatch && dr.MismatchedRows > 0 {
 			mismatch = true
-			log.Infof("Flagging mismatch for %s: %+v", td.table.Name, dr)
+			log.Infof("Flagging mismatch in vdiff %s for %s: %+v", td.wd.ct.uuid, td.table.Name, dr)
 			if err := updateTableMismatch(dbClient, td.wd.ct.id, td.table.Name); err != nil {
 				return nil, err
 			}
 		}
 
 		rowsToCompare--
 		if rowsToCompare < 0 {
-			log.Infof("Stopping vdiff, specified row limit reached")
+			log.Infof("Stopping vdiff %s, specified row limit reached", td.wd.ct.uuid)
 			return dr, nil
 		}
 		if advanceSource {
@@ -889,7 +892,7 @@ func (td *tableDiffer) adjustForSourceTimeZone(targetSelectExprs []sqlparser.Sel
 	if td.wd.ct.sourceTimeZone == "" {
 		return targetSelectExprs
 	}
-	log.Infof("source time zone specified: %s", td.wd.ct.sourceTimeZone)
+	log.Infof("Source time zone specified for vdiff %s: %s", td.wd.ct.uuid, td.wd.ct.sourceTimeZone)
 	var newSelectExprs []sqlparser.SelectExpr
 	var modified bool
 	for _, expr := range targetSelectExprs {
@@ -906,7 +909,7 @@ func (td *tableDiffer) adjustForSourceTimeZone(targetSelectExprs []sqlparser.Sel
 						sqlparser.NewStrLiteral(td.wd.ct.targetTimeZone),
 						sqlparser.NewStrLiteral(td.wd.ct.sourceTimeZone),
 					)
-					log.Infof("converting datetime column %s using convert_tz()", colName)
+					log.Infof("Converting datetime column %s using convert_tz() for vdiff %s", colName, td.wd.ct.uuid)
 					newSelectExprs = append(newSelectExprs, &sqlparser.AliasedExpr{Expr: convertTZFuncExpr, As: colAs.Name})
 					converted = true
 					modified = true
@@ -918,7 +921,8 @@ func (td *tableDiffer) adjustForSourceTimeZone(targetSelectExprs []sqlparser.Sel
 		}
 	}
 	if modified { // at least one datetime was found
-		log.Infof("Found datetime columns when SourceTimeZone was set, resetting target SelectExprs after convert_tz()")
+		log.Infof("Found datetime columns when SourceTimeZone was set, resetting target SelectExprs after convert_tz() for vdiff %s",
+			td.wd.ct.uuid)
 		return newSelectExprs
 	}
 	return targetSelectExprs
@@ -987,11 +991,11 @@ func (td *tableDiffer) getSourcePKCols() error {
 				td.table.Name, topoproto.TabletAliasString(sourceTablet.Tablet.Alias))
 		}
 		if len(pkeCols) > 0 {
-			log.Infof("Using primary key equivalent columns %+v for table %s", pkeCols, td.table.Name)
+			log.Infof("Using primary key equivalent columns %+v for table %s in vdiff %s", pkeCols, td.table.Name, td.wd.ct.uuid)
 			sourceTable.PrimaryKeyColumns = pkeCols
 		} else {
 			// We use every column together as a substitute PK.
-			log.Infof("Using all columns as a substitute primary key for table %s", td.table.Name)
+			log.Infof("Using all columns as a substitute primary key for table %s in vdiff %s", td.table.Name, td.wd.ct.uuid)
 			sourceTable.PrimaryKeyColumns = append(sourceTable.PrimaryKeyColumns, td.table.Columns...)
 		}
 	}

go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go

@@ -172,7 +172,7 @@ func (rs *rowStreamer) buildPlan() error {
 	// filtering will work.
 	rs.plan, err = buildTablePlan(rs.se.Environment(), ti, rs.vschema, rs.query)
 	if err != nil {
-		log.Errorf("%s", err.Error())
+		log.Errorf("Failed to build table plan for %s in row streamer: %s", ti.Name, err.Error())
 		return err
 	}
 
@@ -347,7 +347,7 @@ func (rs *rowStreamer) streamQuery(send func(*binlogdatapb.VStreamRowsResponse)
 		rotatedLog bool
 		err        error
 	)
-	log.Infof("Streaming query: %v\n", rs.sendQuery)
+	log.Infof("Streaming rows for query: %s\n", rs.sendQuery)
 	if rs.mode == RowStreamerModeSingleTable {
 		gtid, rotatedLog, err = rs.conn.streamWithSnapshot(rs.ctx, rs.plan.Table.Name, rs.sendQuery)
 		if err != nil {
@@ -384,7 +384,7 @@ func (rs *rowStreamer) streamQuery(send func(*binlogdatapb.VStreamRowsResponse)
 		Gtid:     gtid,
 	})
 	if err != nil {
-		return fmt.Errorf("stream send error: %v", err)
+		return fmt.Errorf("row stream send error: %v", err)
 	}
 
 	// streamQuery sends heartbeats as long as it operates
@@ -413,7 +413,7 @@ func (rs *rowStreamer) streamQuery(send func(*binlogdatapb.VStreamRowsResponse)
 	logger := logutil.NewThrottledLogger(rs.vse.GetTabletInfo(), throttledLoggerInterval)
 	for {
 		if rs.ctx.Err() != nil {
-			log.Infof("Stream ended because of ctx.Done")
+			log.Infof("RowStreamer stream ended because of ctx.Done")
 			return fmt.Errorf("stream ended: %v", rs.ctx.Err())
 		}
 
@@ -422,7 +422,7 @@ func (rs *rowStreamer) streamQuery(send func(*binlogdatapb.VStreamRowsResponse)
 			throttleResponseRateLimiter.Do(func() error {
 				return safeSend(&binlogdatapb.VStreamRowsResponse{Throttled: true, ThrottledReason: checkResult.Summary()})
 			})
-			logger.Infof("throttled.")
+			logger.Infof("Throttled streaming rows for %s", rs.sendQuery)
 			continue
 		}