test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-11-30 02:48:45 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Make progress if any channel/segment was loaded on node (#20775)

Browse Source 
Signed-off-by: yah01 <yang.cen@zilliz.com>

Signed-off-by: yah01 <yang.cen@zilliz.com>
This commit is contained in:
yah01 2022-12-06 18:29:19 +08:00 committed by GitHub
parent 11e4445ef7
commit 0297ab1a46
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 30 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/querycoordv2/meta/leader_view_manager.go
View File

@ -183,7 +183,7 @@ func (mgr *LeaderViewManager) GetGrowingSegmentDistByCollectionAndNode(collectio
return segments
}
// GetSegmentDist returns the list of nodes the given segment on
// GetSegmentDist returns the list of nodes the given channel on
func (mgr *LeaderViewManager) GetChannelDist(channel string) []int64 {
mgr.rwmutex.RLock()
defer mgr.rwmutex.RUnlock()

32
internal/querycoordv2/observers/collection_observer.go
View File

@ -166,7 +166,9 @@ func (ob *CollectionObserver) observeCollectionLoadStatus(collection *meta.Colle
log.Info("collection targets",
zap.Int("segmentTargetNum", len(segmentTargets)),
zap.Int("channelTargetNum", len(channelTargets)),
zap.Int("totalTargetNum", targetNum))
zap.Int("totalTargetNum", targetNum),
zap.Int32("replicaNum", collection.GetReplicaNumber()),
)
updated := collection.Clone()
loadedCount := 0
@ -178,18 +180,14 @@ func (ob *CollectionObserver) observeCollectionLoadStatus(collection *meta.Colle
group := utils.GroupNodesByReplica(ob.meta.ReplicaManager,
collection.GetCollectionID(),
ob.dist.LeaderViewManager.GetChannelDist(channel.GetChannelName()))
if len(group) >= int(collection.GetReplicaNumber()) {
loadedCount++
}
loadedCount += len(group)
}
subChannelCount := loadedCount
for _, segment := range segmentTargets {
group := utils.GroupNodesByReplica(ob.meta.ReplicaManager,
collection.GetCollectionID(),
ob.dist.LeaderViewManager.GetSealedSegmentDist(segment.GetID()))
if len(group) >= int(collection.GetReplicaNumber()) {
loadedCount++
}
loadedCount += len(group)
}
if loadedCount > 0 {
log.Info("collection load progress",
@ -198,7 +196,7 @@ func (ob *CollectionObserver) observeCollectionLoadStatus(collection *meta.Colle
)
}
updated.LoadPercentage = int32(loadedCount * 100 / targetNum)
updated.LoadPercentage = int32(loadedCount * 100 / targetNum * int(collection.GetReplicaNumber()))
}
if loadedCount <= ob.collectionLoadedCount[collection.GetCollectionID()] && updated.LoadPercentage != 100 {
@ -230,9 +228,11 @@ func (ob *CollectionObserver) observePartitionLoadStatus(partition *meta.Partiti
channelTargets := ob.targetMgr.GetDmChannelsByCollection(partition.GetCollectionID(), meta.NextTarget)
targetNum := len(segmentTargets) + len(channelTargets)
log.Info("partition targets",
zap.Int("segment-target-num", len(segmentTargets)),
zap.Int("channel-target-num", len(channelTargets)),
zap.Int("total-target-num", targetNum))
zap.Int("segmentTargetNum", len(segmentTargets)),
zap.Int("channelTargetNum", len(channelTargets)),
zap.Int("totalTargetNum", targetNum),
zap.Int32("replicaNum", partition.GetReplicaNumber()),
)
loadedCount := 0
updated := partition.Clone()
@ -244,25 +244,21 @@ func (ob *CollectionObserver) observePartitionLoadStatus(partition *meta.Partiti
group := utils.GroupNodesByReplica(ob.meta.ReplicaManager,
partition.GetCollectionID(),
ob.dist.LeaderViewManager.GetChannelDist(channel.GetChannelName()))
if len(group) >= int(partition.GetReplicaNumber()) {
loadedCount++
}
loadedCount += len(group)
}
subChannelCount := loadedCount
for _, segment := range segmentTargets {
group := utils.GroupNodesByReplica(ob.meta.ReplicaManager,
partition.GetCollectionID(),
ob.dist.LeaderViewManager.GetSealedSegmentDist(segment.GetID()))
if len(group) >= int(partition.GetReplicaNumber()) {
loadedCount++
}
loadedCount += len(group)
}
if loadedCount > 0 {
log.Info("partition load progress",
zap.Int("subChannelCount", subChannelCount),
zap.Int("loadSegmentCount", loadedCount-subChannelCount))
}
updated.LoadPercentage = int32(loadedCount * 100 / targetNum)
updated.LoadPercentage = int32(loadedCount * 100 / targetNum * int(partition.GetReplicaNumber()))
}

17
internal/querycoordv2/observers/collection_observer_test.go
View File

@ -199,6 +199,15 @@ func (suite *CollectionObserverSuite) TestObserve() {
time := suite.meta.GetCollection(suite.collections[2]).UpdatedAt
// Not timeout
Params.QueryCoordCfg.LoadTimeoutSeconds = timeout
segments := []*datapb.SegmentBinlogs{}
for _, segment := range suite.segments[100] {
segments = append(segments, &datapb.SegmentBinlogs{
SegmentID: segment.GetID(),
InsertChannel: segment.GetInsertChannel(),
})
}
suite.ob.Start(context.Background())
// Collection 100 loaded before timeout,
@ -325,8 +334,12 @@ func (suite *CollectionObserverSuite) load(collection int64) {
})
}
suite.broker.EXPECT().GetRecoveryInfo(mock.Anything, collection, int64(1)).Return(dmChannels, allSegments, nil)
suite.targetMgr.UpdateCollectionNextTargetWithPartitions(collection, int64(1))
partitions := suite.partitions[collection]
for _, partition := range partitions {
suite.broker.EXPECT().GetRecoveryInfo(mock.Anything, collection, partition).Return(dmChannels, allSegments, nil)
}
suite.targetMgr.UpdateCollectionNextTargetWithPartitions(collection, partitions...)
}
func TestCollectionObserver(t *testing.T) {