Use tickle to check WatchEvent state at datacoord and datanode

configs/milvus.yaml

@@ -280,7 +280,7 @@ dataCoord:
   enableActiveStandby: false  # Enable active-standby
 
   channel:
-    maxWatchDuration: 600 # Timeout on watching channels (in seconds). Default 600 seconds.
+    watchTimeoutInterval: 30 # Timeout on watching channels (in seconds). Datanode tickler update watch progress will reset timeout timer.
 
   segment:
     maxSize: 512 # Maximum size of a segment in MB

internal/datacoord/channel_checker.go

@@ -33,10 +33,12 @@ import (
 )
 
 type channelStateTimer struct {
-	watchkv        kv.MetaKv
-	runningTimers  sync.Map // channel name to timer stop channels
-	etcdWatcher    clientv3.WatchChan
-	timeoutWatcher chan *ackEvent
+	watchkv kv.MetaKv
+
+	runningTimers     sync.Map
+	runningTimerStops sync.Map // channel name to timer stop channels
+	etcdWatcher       clientv3.WatchChan
+	timeoutWatcher    chan *ackEvent
 }
 
 func newChannelStateTimer(kv kv.MetaKv) *channelStateTimer {
@@ -49,7 +51,6 @@ func newChannelStateTimer(kv kv.MetaKv) *channelStateTimer {
 func (c *channelStateTimer) getWatchers(prefix string) (clientv3.WatchChan, chan *ackEvent) {
 	if c.etcdWatcher == nil {
 		c.etcdWatcher = c.watchkv.WatchWithPrefix(prefix)
-
 	}
 	return c.etcdWatcher, c.timeoutWatcher
 }
@@ -80,40 +81,46 @@ func (c *channelStateTimer) loadAllChannels(nodeID UniqueID) ([]*datapb.ChannelW
 }
 
 // startOne can write ToWatch or ToRelease states.
-func (c *channelStateTimer) startOne(watchState datapb.ChannelWatchState, channelName string, nodeID UniqueID, timeoutTs int64) {
-	if timeoutTs == 0 {
+func (c *channelStateTimer) startOne(watchState datapb.ChannelWatchState, channelName string, nodeID UniqueID, timeout time.Duration) {
+	if timeout == 0 {
 		log.Info("zero timeoutTs, skip starting timer",
 			zap.String("watch state", watchState.String()),
 			zap.Int64("nodeID", nodeID),
 			zap.String("channel name", channelName),
 		)
 		return
 	}
+
 	stop := make(chan struct{})
+	ticker := time.NewTimer(timeout)
 	c.removeTimers([]string{channelName})
-	c.runningTimers.Store(channelName, stop)
-	timeoutT := time.Unix(0, timeoutTs)
+	c.runningTimerStops.Store(channelName, stop)
+	c.runningTimers.Store(channelName, ticker)
+
 	go func() {
 		log.Info("timer started",
 			zap.String("watch state", watchState.String()),
 			zap.Int64("nodeID", nodeID),
 			zap.String("channel name", channelName),
-			zap.Time("timeout time", timeoutT))
+			zap.Duration("check interval", timeout))
+		defer ticker.Stop()
+
 		select {
-		case <-time.NewTimer(time.Until(timeoutT)).C:
+		case <-ticker.C:
+			// check tickle at path as :tickle/[prefix]/{channel_name}
 			log.Info("timeout and stop timer: wait for channel ACK timeout",
 				zap.String("watch state", watchState.String()),
 				zap.Int64("nodeID", nodeID),
 				zap.String("channel name", channelName),
-				zap.Time("timeout time", timeoutT))
+				zap.Duration("timeout interval", timeout))
 			ackType := getAckType(watchState)
 			c.notifyTimeoutWatcher(&ackEvent{ackType, channelName, nodeID})
 		case <-stop:
 			log.Info("stop timer before timeout",
 				zap.String("watch state", watchState.String()),
 				zap.Int64("nodeID", nodeID),
 				zap.String("channel name", channelName),
-				zap.Time("timeout time", timeoutT))
+				zap.Duration("timeout interval", timeout))
 		}
 	}()
 }
@@ -124,16 +131,25 @@ func (c *channelStateTimer) notifyTimeoutWatcher(e *ackEvent) {
 
 func (c *channelStateTimer) removeTimers(channels []string) {
 	for _, channel := range channels {
-		if stop, ok := c.runningTimers.LoadAndDelete(channel); ok {
+		if stop, ok := c.runningTimerStops.LoadAndDelete(channel); ok {
 			close(stop.(chan struct{}))
+			c.runningTimers.Delete(channel)
 		}
 	}
 }
 
 func (c *channelStateTimer) stopIfExist(e *ackEvent) {
-	stop, ok := c.runningTimers.LoadAndDelete(e.channelName)
+	stop, ok := c.runningTimerStops.LoadAndDelete(e.channelName)
 	if ok && e.ackType != watchTimeoutAck && e.ackType != releaseTimeoutAck {
 		close(stop.(chan struct{}))
+		c.runningTimers.Delete(e.channelName)
+	}
+}
+
+func (c *channelStateTimer) resetIfExist(channel string, interval time.Duration) {
+	if value, ok := c.runningTimers.Load(channel); ok {
+		timer := value.(*time.Timer)
+		timer.Reset(interval)
 	}
 }
 

internal/datacoord/channel_checker_test.go

@@ -52,10 +52,9 @@ func TestChannelStateTimer(t *testing.T) {
 		timer.loadAllChannels(1)
 
 		validWatchInfo := datapb.ChannelWatchInfo{
-			Vchan:     &datapb.VchannelInfo{},
-			StartTs:   time.Now().Unix(),
-			State:     datapb.ChannelWatchState_ToWatch,
-			TimeoutTs: time.Now().Add(20 * time.Millisecond).UnixNano(),
+			Vchan:   &datapb.VchannelInfo{},
+			StartTs: time.Now().Unix(),
+			State:   datapb.ChannelWatchState_ToWatch,
 		}
 		validData, err := proto.Marshal(&validWatchInfo)
 		require.NoError(t, err)
@@ -87,18 +86,20 @@ func TestChannelStateTimer(t *testing.T) {
 	})
 
 	t.Run("test startOne", func(t *testing.T) {
-		normalTimeoutTs := time.Now().Add(20 * time.Second).UnixNano()
-		nowTimeoutTs := time.Now().UnixNano()
-		zeroTimeoutTs := int64(0)
+		normalTimeoutTs := 20 * time.Second
+		nowTimeoutTs := 1 * time.Millisecond
+		zeroTimeoutTs := 0 * time.Second
+		resetTimeoutTs := 30 * time.Second
 		tests := []struct {
 			channelName string
-			timeoutTs   int64
+			timeoutTs   time.Duration
 
 			description string
 		}{
 			{"channel-1", normalTimeoutTs, "test stop"},
 			{"channel-2", nowTimeoutTs, "test timeout"},
 			{"channel-3", zeroTimeoutTs, "not start"},
+			{"channel-4", resetTimeoutTs, "reset timer"},
 		}
 
 		timer := newChannelStateTimer(kv)
@@ -112,6 +113,11 @@ func TestChannelStateTimer(t *testing.T) {
 					e := <-timeoutCh
 					assert.Equal(t, watchTimeoutAck, e.ackType)
 					assert.Equal(t, test.channelName, e.channelName)
+				} else if test.timeoutTs == resetTimeoutTs {
+					timer.resetIfExist(test.channelName, nowTimeoutTs)
+					e := <-timeoutCh
+					assert.Equal(t, watchTimeoutAck, e.ackType)
+					assert.Equal(t, test.channelName, e.channelName)
 				} else {
 					timer.stopIfExist(&ackEvent{watchSuccessAck, test.channelName, 1})
 				}
@@ -123,18 +129,17 @@ func TestChannelStateTimer(t *testing.T) {
 	})
 
 	t.Run("test startOne no leaking issue 17335", func(t *testing.T) {
-		timeoutTs := time.Now().Add(20 * time.Second).UnixNano()
 		timer := newChannelStateTimer(kv)
 
-		timer.startOne(datapb.ChannelWatchState_ToRelease, "channel-1", 1, timeoutTs)
-		stop, ok := timer.runningTimers.Load("channel-1")
+		timer.startOne(datapb.ChannelWatchState_ToRelease, "channel-1", 1, 20*time.Second)
+		stop, ok := timer.runningTimerStops.Load("channel-1")
 		require.True(t, ok)
 
-		timer.startOne(datapb.ChannelWatchState_ToWatch, "channel-1", 1, timeoutTs)
+		timer.startOne(datapb.ChannelWatchState_ToWatch, "channel-1", 1, 20*time.Second)
 		_, ok = <-stop.(chan struct{})
 		assert.False(t, ok)
 
-		stop2, ok := timer.runningTimers.Load("channel-1")
+		stop2, ok := timer.runningTimerStops.Load("channel-1")
 		assert.True(t, ok)
 
 		timer.removeTimers([]string{"channel-1"})
@@ -151,10 +156,9 @@ func TestChannelStateTimer_parses(t *testing.T) {
 
 	t.Run("test parseWatchInfo", func(t *testing.T) {
 		validWatchInfo := datapb.ChannelWatchInfo{
-			Vchan:     &datapb.VchannelInfo{},
-			StartTs:   time.Now().Unix(),
-			State:     datapb.ChannelWatchState_ToWatch,
-			TimeoutTs: time.Now().Add(20 * time.Millisecond).UnixNano(),
+			Vchan:   &datapb.VchannelInfo{},
+			StartTs: time.Now().Unix(),
+			State:   datapb.ChannelWatchState_ToWatch,
 		}
 		validData, err := proto.Marshal(&validWatchInfo)
 		require.NoError(t, err)
@@ -186,7 +190,6 @@ func TestChannelStateTimer_parses(t *testing.T) {
 					assert.NotNil(t, info)
 					assert.Equal(t, info.GetState(), validWatchInfo.GetState())
 					assert.Equal(t, info.GetStartTs(), validWatchInfo.GetStartTs())
-					assert.Equal(t, info.GetTimeoutTs(), validWatchInfo.GetTimeoutTs())
 				} else {
 					assert.Nil(t, info)
 					assert.Error(t, err)
@@ -205,9 +208,8 @@ func TestChannelStateTimer_parses(t *testing.T) {
 				DroppedSegments:     []*datapb.SegmentInfo{{ID: 3}},
 				UnflushedSegmentIds: []int64{1},
 			},
-			StartTs:   time.Now().Unix(),
-			State:     datapb.ChannelWatchState_ToWatch,
-			TimeoutTs: time.Now().Add(20 * time.Millisecond).UnixNano(),
+			StartTs: time.Now().Unix(),
+			State:   datapb.ChannelWatchState_ToWatch,
 		}
 
 		oldData, err := proto.Marshal(&oldWatchInfo)

internal/datacoord/channel_manager.go

@@ -192,22 +192,23 @@ func (c *ChannelManager) checkOldNodes(nodes []UniqueID) error {
 	for nodeID, watchInfos := range nodeWatchInfos {
 		for _, info := range watchInfos {
 			channelName := info.GetVchan().GetChannelName()
+			checkInterval := Params.DataCoordCfg.WatchTimeoutInterval.GetAsDuration(time.Second)
 
 			log.Info("processing watch info",
 				zap.String("watch state", info.GetState().String()),
 				zap.String("channel name", channelName))
 
 			switch info.GetState() {
 			case datapb.ChannelWatchState_ToWatch, datapb.ChannelWatchState_Uncomplete:
-				c.stateTimer.startOne(datapb.ChannelWatchState_ToWatch, channelName, nodeID, info.GetTimeoutTs())
+				c.stateTimer.startOne(datapb.ChannelWatchState_ToWatch, channelName, nodeID, checkInterval)
 
 			case datapb.ChannelWatchState_WatchFailure:
 				if err := c.Release(nodeID, channelName); err != nil {
 					return err
 				}
 
 			case datapb.ChannelWatchState_ToRelease:
-				c.stateTimer.startOne(datapb.ChannelWatchState_ToRelease, channelName, nodeID, info.GetTimeoutTs())
+				c.stateTimer.startOne(datapb.ChannelWatchState_ToRelease, channelName, nodeID, checkInterval)
 
 			case datapb.ChannelWatchState_ReleaseSuccess:
 				if err := c.Reassign(nodeID, channelName); err != nil {
@@ -446,11 +447,10 @@ func (c *ChannelManager) fillChannelWatchInfo(op *ChannelOp) {
 	for _, ch := range op.Channels {
 		vcInfo := c.h.GetDataVChanPositions(ch, allPartitionID)
 		info := &datapb.ChannelWatchInfo{
-			Vchan:     vcInfo,
-			StartTs:   time.Now().Unix(),
-			State:     datapb.ChannelWatchState_Uncomplete,
-			TimeoutTs: time.Now().Add(Params.DataCoordCfg.MaxWatchDuration.GetAsDuration(time.Second)).UnixNano(),
-			Schema:    ch.Schema,
+			Vchan:   vcInfo,
+			StartTs: time.Now().Unix(),
+			State:   datapb.ChannelWatchState_Uncomplete,
+			Schema:  ch.Schema,
 		}
 		op.ChannelWatchInfos = append(op.ChannelWatchInfos, info)
 	}
@@ -460,20 +460,19 @@ func (c *ChannelManager) fillChannelWatchInfo(op *ChannelOp) {
 func (c *ChannelManager) fillChannelWatchInfoWithState(op *ChannelOp, state datapb.ChannelWatchState) []string {
 	var channelsWithTimer = []string{}
 	startTs := time.Now().Unix()
-	timeoutTs := time.Now().Add(Params.DataCoordCfg.MaxWatchDuration.GetAsDuration(time.Second)).UnixNano()
+	checkInterval := Params.DataCoordCfg.WatchTimeoutInterval.GetAsDuration(time.Second)
 	for _, ch := range op.Channels {
 		vcInfo := c.h.GetDataVChanPositions(ch, allPartitionID)
 		info := &datapb.ChannelWatchInfo{
-			Vchan:     vcInfo,
-			StartTs:   startTs,
-			State:     state,
-			TimeoutTs: timeoutTs,
-			Schema:    ch.Schema,
+			Vchan:   vcInfo,
+			StartTs: startTs,
+			State:   state,
+			Schema:  ch.Schema,
 		}
 
 		// Only set timer for watchInfo not from bufferID
 		if op.NodeID != bufferID {
-			c.stateTimer.startOne(state, ch.Name, op.NodeID, timeoutTs)
+			c.stateTimer.startOne(state, ch.Name, op.NodeID, checkInterval)
 			channelsWithTimer = append(channelsWithTimer, ch.Name)
 		}
 
@@ -696,11 +695,13 @@ func (c *ChannelManager) watchChannelStatesLoop(ctx context.Context) {
 					continue
 				}
 
-				// ignore these states
+				// runnging states
 				state := watchInfo.GetState()
 				if state == datapb.ChannelWatchState_ToWatch ||
 					state == datapb.ChannelWatchState_ToRelease ||
 					state == datapb.ChannelWatchState_Uncomplete {
+					c.stateTimer.resetIfExist(watchInfo.GetVchan().ChannelName, Params.DataCoordCfg.WatchTimeoutInterval.GetAsDuration(time.Second))
+					log.Info("tickle update, timer delay", zap.String("channel", watchInfo.GetVchan().ChannelName), zap.Int32("progress", watchInfo.Progress))
 					continue
 				}