kvclient(ticdc): add table id to kv client logs

cdc/kv/region_state.go

@@ -45,7 +45,7 @@ type regionInfo struct {
 	lockedRangeState *regionlock.LockedRangeState
 }
 
-func (s regionInfo) isStoped() bool {
+func (s regionInfo) isStopped() bool {
 	// lockedRange only nil when the region's subscribedTable is stopped.
 	return s.lockedRangeState == nil
 }

cdc/kv/shared_client.go

@@ -288,13 +288,14 @@
 	s.totalSpans.Lock()
 	s.totalSpans.v[subID] = rt
 	s.totalSpans.Unlock()
-
 	s.rangeTaskCh.In() <- rangeTask{span: span, subscribedTable: rt}
 	log.Info("event feed subscribes table success",
 		zap.String("namespace", s.changefeed.Namespace),
 		zap.String("changefeed", s.changefeed.ID),
 		zap.Any("subscriptionID", rt.subscriptionID),
-		zap.String("span", rt.span.String()))
+		zap.Int64("tableID", rt.span.TableID),
+		zap.Any("startKey", rt.span.StartKey),
+		zap.Any("endKey", rt.span.EndKey))
 }
 
 // Unsubscribe the given table span. All covered regions will be deregistered asynchronously.
@@ -306,13 +307,19 @@
 	s.totalSpans.Unlock()
 	if rt != nil {
 		s.setTableStopped(rt)
+		log.Info("event feed unsubscribes table",
+			zap.String("namespace", s.changefeed.Namespace),
+			zap.String("changefeed", s.changefeed.ID),
+			zap.Any("subscriptionID", rt.subscriptionID),
+			zap.Int64("tableID", rt.span.TableID),
+			zap.Any("startKey", rt.span.StartKey),
+			zap.Any("endKey", rt.span.EndKey))
+		return
 	}
-
-	log.Info("event feed unsubscribes table",
+	log.Warn("event feed unsubscribes table, but not found",
 		zap.String("namespace", s.changefeed.Namespace),
 		zap.String("changefeed", s.changefeed.ID),
-		zap.Any("subscriptionID", rt.subscriptionID),
-		zap.Bool("exists", rt != nil))
+		zap.Any("subscriptionID", subID))
 }
 
 // ResolveLock is a function. If outsider subscribers find a span resolved timestamp is
@@ -382,7 +389,8 @@
 	log.Info("event feed starts to stop table",
 		zap.String("namespace", s.changefeed.Namespace),
 		zap.String("changefeed", s.changefeed.ID),
-		zap.Any("subscriptionID", rt.subscriptionID))
+		zap.Any("subscriptionID", rt.subscriptionID),
+		zap.Int64("tableID", rt.span.TableID))
 
 	// Set stopped to true so we can stop handling region events from the table.
 	// Then send a special singleRegionInfo to regionRouter to deregister the table
@@ -399,7 +407,8 @@
 	log.Info("event feed stop table is finished",
 		zap.String("namespace", s.changefeed.Namespace),
 		zap.String("changefeed", s.changefeed.ID),
-		zap.Any("subscriptionID", rt.subscriptionID))
+		zap.Any("subscriptionID", rt.subscriptionID),
+		zap.Int64("tableID", rt.span.TableID))
 
 	s.totalSpans.Lock()
 	defer s.totalSpans.Unlock()
@@ -418,7 +427,7 @@
 		case <-ctx.Done():
 			return errors.Trace(ctx.Err())
 		case region := <-s.regionCh.Out():
-			if region.isStoped() {
+			if region.isStopped() {
 				for _, rs := range s.stores {
 					s.broadcastRequest(rs, region)
 				}
@@ -440,7 +449,10 @@
 				zap.String("changefeed", s.changefeed.ID),
 				zap.Uint64("streamID", stream.streamID),
 				zap.Any("subscriptionID", region.subscribedTable.subscriptionID),
+				zap.Int64("tableID", region.span.TableID),
 				zap.Uint64("regionID", region.verID.GetID()),
+				zap.Any("startKey", region.span.StartKey),
+				zap.Any("endKey", region.span.EndKey),
 				zap.Uint64("storeID", store.storeID),
 				zap.String("addr", store.storeAddr))
 		}
@@ -554,7 +566,9 @@
 				zap.String("namespace", s.changefeed.Namespace),
 				zap.String("changefeed", s.changefeed.ID),
 				zap.Any("subscriptionID", subscribedTable.subscriptionID),
-				zap.Any("span", nextSpan),
+				zap.Int64("tableID", nextSpan.TableID),
+				zap.Any("startKey", nextSpan.StartKey),
+				zap.Any("endKey", nextSpan.EndKey),
 				zap.Error(err))
 			backoffBeforeLoad = true
 			continue
@@ -572,7 +586,9 @@
 				zap.String("namespace", s.changefeed.Namespace),
 				zap.String("changefeed", s.changefeed.ID),
 				zap.Any("subscriptionID", subscribedTable.subscriptionID),
-				zap.Any("span", nextSpan))
+				zap.Int64("tableID", nextSpan.TableID),
+				zap.Any("startKey", nextSpan.StartKey),
+				zap.Any("endKey", nextSpan.EndKey))
 			backoffBeforeLoad = true
 			continue
 		}
@@ -590,7 +606,10 @@
 				log.Panic("event feed check spans intersect shouldn't fail",
 					zap.String("namespace", s.changefeed.Namespace),
 					zap.String("changefeed", s.changefeed.ID),
-					zap.Any("subscriptionID", subscribedTable.subscriptionID))
+					zap.Any("subscriptionID", subscribedTable.subscriptionID),
+					zap.Int64("tableID", nextSpan.TableID),
+					zap.Any("startKey", nextSpan.StartKey),
+					zap.Any("endKey", nextSpan.EndKey))
 			}
 
 			verID := tikv.NewRegionVerID(regionMeta.Id, regionMeta.RegionEpoch.ConfVer, regionMeta.RegionEpoch.Version)
@@ -676,6 +695,8 @@
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
 			zap.Any("subscriptionID", errInfo.subscribedTable.subscriptionID),
+			zap.Uint64("regionID", errInfo.verID.GetID()),
+			zap.Int64("tableID", errInfo.span.TableID),
 			zap.Stringer("error", innerErr))
 
 		if notLeader := innerErr.GetNotLeader(); notLeader != nil {
@@ -720,6 +741,8 @@
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
 			zap.Any("subscriptionID", errInfo.subscribedTable.subscriptionID),
+			zap.Uint64("regionID", errInfo.verID.GetID()),
+			zap.Int64("tableID", errInfo.span.TableID),
 			zap.Stringer("error", innerErr))
 		metricFeedUnknownErrorCounter.Inc()
 		s.scheduleRegionRequest(ctx, errInfo.regionInfo)
@@ -740,6 +763,8 @@
 			zap.String("namespace", s.changefeed.Namespace),
 			zap.String("changefeed", s.changefeed.ID),
 			zap.Any("subscriptionID", errInfo.subscribedTable.subscriptionID),
+			zap.Uint64("regionID", errInfo.verID.GetID()),
+			zap.Int64("tableID", errInfo.span.TableID),
 			zap.Error(err))
 		return err
 	}
@@ -816,7 +841,7 @@
 
 		currTime := s.pdClock.CurrentTime()
 		s.totalSpans.RLock()
-		slowInitializeRegion := 0
+		var slowInitializeRegionCount int
 		for subscriptionID, rt := range s.totalSpans.v {
 			attr := rt.rangeLock.IterAll(nil)
 			ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.ResolvedTs)
@@ -826,32 +851,36 @@
 						zap.String("namespace", s.changefeed.Namespace),
 						zap.String("changefeed", s.changefeed.ID),
 						zap.Any("subscriptionID", subscriptionID),
+						zap.Int64("tableID", rt.span.TableID),
 						zap.Any("slowRegion", attr.SlowestRegion))
 				}
 			} else if currTime.Sub(attr.SlowestRegion.Created) > 10*time.Minute {
-				slowInitializeRegion += 1
+				slowInitializeRegionCount += 1
 				log.Info("event feed initializes a region too slow",
 					zap.String("namespace", s.changefeed.Namespace),
 					zap.String("changefeed", s.changefeed.ID),
 					zap.Any("subscriptionID", subscriptionID),
+					zap.Int64("tableID", rt.span.TableID),
 					zap.Any("slowRegion", attr.SlowestRegion))
 			} else if currTime.Sub(ckptTime) > 10*time.Minute {
 				log.Info("event feed finds a uninitialized slow region",
 					zap.String("namespace", s.changefeed.Namespace),
 					zap.String("changefeed", s.changefeed.ID),
 					zap.Any("subscriptionID", subscriptionID),
+					zap.Int64("tableID", rt.span.TableID),
 					zap.Any("slowRegion", attr.SlowestRegion))
 			}
 			if len(attr.UnLockedRanges) > 0 {
 				log.Info("event feed holes exist",
 					zap.String("namespace", s.changefeed.Namespace),
 					zap.String("changefeed", s.changefeed.ID),
 					zap.Any("subscriptionID", subscriptionID),
+					zap.Int64("tableID", rt.span.TableID),
 					zap.Any("holes", attr.UnLockedRanges))
 			}
 		}
 		s.totalSpans.RUnlock()
-		s.metrics.slowInitializeRegion.Set(float64(slowInitializeRegion))
+		s.metrics.slowInitializeRegion.Set(float64(slowInitializeRegionCount))
 	}
 }
 

cdc/kv/shared_region_worker.go

@@ -175,6 +175,7 @@ func (w *sharedRegionWorker) handleSingleRegionError(state *regionFeedState, str
 			zap.Uint64("streamID", stream.streamID),
 			zap.Any("subscriptionID", state.getRegionID()),
 			zap.Uint64("regionID", state.region.verID.GetID()),
+			zap.Int64("tableID", state.region.span.TableID),
 			zap.Bool("reschedule", stepsToRemoved),
 			zap.Error(err))
 	}
@@ -228,12 +229,6 @@ func (w *sharedRegionWorker) handleEventEntry(ctx context.Context, x *cdcpb.Even
 		}
 	}
 	tableID := state.region.subscribedTable.span.TableID
-	log.Debug("region worker get an Event",
-		zap.String("namespace", w.changefeed.Namespace),
-		zap.String("changefeed", w.changefeed.ID),
-		zap.Any("subscriptionID", state.region.subscribedTable.subscriptionID),
-		zap.Int64("tableID", tableID),
-		zap.Int("rows", len(x.Entries.GetEntries())))
 	return handleEventEntry(x, startTs, state, w.metrics, emit, w.changefeed, tableID, w.client.logRegionDetails)
 }
 
@@ -250,7 +245,7 @@ func handleEventEntry(
 	regionID, regionSpan, _ := state.getRegionMeta()
 	for _, entry := range x.Entries.GetEntries() {
 		// NOTE: from TiKV 7.0.0, entries are already filtered out in TiKV side.
-		// We can remove the check in future.
+		// We can remove the check in the future.
 		comparableKey := spanz.ToComparableKey(entry.GetKey())
 		if entry.Type != cdcpb.Event_INITIALIZED &&
 			!spanz.KeyInSpan(comparableKey, regionSpan) {
@@ -266,6 +261,7 @@ func handleEventEntry(
 				zap.String("changefeed", changefeed.ID),
 				zap.Int64("tableID", tableID),
 				zap.Uint64("regionID", regionID),
+				zap.Int64("tableID", state.region.span.TableID),
 				zap.Uint64("requestID", state.requestID),
 				zap.Stringer("span", &state.region.span))
 

cdc/kv/shared_stream.go

@@ -76,7 +76,7 @@
 			case <-ctx.Done():
 				return ctx.Err()
 			case region := <-stream.requests.Out():
-				if !region.isStoped() {
+				if !region.isStopped() {
 					stream.preFetchForConnecting = new(regionInfo)
 					*stream.preFetchForConnecting = region
 					return nil
@@ -104,7 +104,7 @@
 			// Why we need to re-schedule pending regions? This because the store can
 			// fail forever, and all regions are scheduled to other stores.
 			for _, region := range stream.clearPendingRegions() {
-				if region.isStoped() {
+				if region.isStopped() {
 					// It means it's a special task for stopping the table.
 					continue
 				}
@@ -254,30 +254,6 @@
 }
 
 func (s *requestedStream) send(ctx context.Context, c *SharedClient, rs *requestedStore) (err error) {
-	doSend := func(cc *sharedconn.ConnAndClient, req *cdcpb.ChangeDataRequest, subscriptionID SubscriptionID) error {
-		if err := cc.Client().Send(req); err != nil {
-			log.Warn("event feed send request to grpc stream failed",
-				zap.String("namespace", c.changefeed.Namespace),
-				zap.String("changefeed", c.changefeed.ID),
-				zap.Uint64("streamID", s.streamID),
-				zap.Any("subscriptionID", subscriptionID),
-				zap.Uint64("regionID", req.RegionId),
-				zap.Uint64("storeID", rs.storeID),
-				zap.String("addr", rs.storeAddr),
-				zap.Error(err))
-			return errors.Trace(err)
-		}
-		log.Debug("event feed send request to grpc stream success",
-			zap.String("namespace", c.changefeed.Namespace),
-			zap.String("changefeed", c.changefeed.ID),
-			zap.Uint64("streamID", s.streamID),
-			zap.Any("subscriptionID", subscriptionID),
-			zap.Uint64("regionID", req.RegionId),
-			zap.Uint64("storeID", rs.storeID),
-			zap.String("addr", rs.storeAddr))
-		return nil
-	}
-
 	fetchMoreReq := func() (regionInfo, error) {
 		waitReqTicker := time.NewTicker(60 * time.Second)
 		defer waitReqTicker.Stop()
@@ -329,23 +305,24 @@
 	s.preFetchForConnecting = nil
 	for {
 		subscriptionID := region.subscribedTable.subscriptionID
-		log.Debug("event feed gets a singleRegionInfo",
-			zap.String("namespace", c.changefeed.Namespace),
-			zap.String("changefeed", c.changefeed.ID),
-			zap.Uint64("streamID", s.streamID),
-			zap.Any("subscriptionID", subscriptionID),
-			zap.Uint64("regionID", region.verID.GetID()),
-			zap.Uint64("storeID", rs.storeID),
-			zap.String("addr", rs.storeAddr))
 		// It means it's a special task for stopping the table.
-		if region.isStoped() {
+		if region.isStopped() {
 			if s.multiplexing != nil {
 				req := &cdcpb.ChangeDataRequest{
 					RequestId: uint64(subscriptionID),
 					Request:   &cdcpb.ChangeDataRequest_Deregister_{},
 				}
-				if err = doSend(s.multiplexing, req, subscriptionID); err != nil {
-					return err
+				if err = s.multiplexing.Client().Send(req); err != nil {
+					log.Warn("event feed send deregister request to grpc stream failed",
+						zap.String("namespace", c.changefeed.Namespace),
+						zap.String("changefeed", c.changefeed.ID),
+						zap.Uint64("streamID", s.streamID),
+						zap.Any("subscriptionID", subscriptionID),
+						zap.Int64("tableID", region.span.TableID),
+						zap.Uint64("regionID", req.RegionId),
+						zap.Uint64("storeID", rs.storeID),
+						zap.String("addr", rs.storeAddr),
+						zap.Error(err))
 				}
 			} else if cc := tableExclusives[subscriptionID]; cc != nil {
 				delete(tableExclusives, subscriptionID)
@@ -385,8 +362,17 @@
 			} else if cc, err = getTableExclusiveConn(subscriptionID); err != nil {
 				return err
 			}
-			if err = doSend(cc, c.createRegionRequest(region), subscriptionID); err != nil {
-				return err
+			if err = cc.Client().Send(c.createRegionRequest(region)); err != nil {
+				log.Warn("event feed send request to grpc stream failed",
+					zap.String("namespace", c.changefeed.Namespace),
+					zap.String("changefeed", c.changefeed.ID),
+					zap.Uint64("streamID", s.streamID),
+					zap.Any("subscriptionID", subscriptionID),
+					zap.Uint64("regionID", region.verID.GetID()),
+					zap.Int64("tableID", region.span.TableID),
+					zap.Uint64("storeID", rs.storeID),
+					zap.String("addr", rs.storeAddr),
+					zap.Error(err))
 			}
 		}
 
@@ -483,14 +469,19 @@
 		state := s.getState(subscriptionID, regionID)
 		switch x := event.Event.(type) {
 		case *cdcpb.Event_Error:
-			s.logRegionDetails("event feed receives a region error",
+			fields := []zap.Field{
 				zap.String("namespace", c.changefeed.Namespace),
 				zap.String("changefeed", c.changefeed.ID),
 				zap.Uint64("streamID", s.streamID),
 				zap.Any("subscriptionID", subscriptionID),
 				zap.Uint64("regionID", event.RegionId),
 				zap.Bool("stateIsNil", state == nil),
-				zap.Any("error", x.Error))
+				zap.Any("error", x.Error),
+			}
+			if state != nil {
+				fields = append(fields, zap.Int64("tableID", state.region.span.TableID))
+			}
+			s.logRegionDetails("event feed receives a region error", fields...)
 		}
 
 		if state != nil {