test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-12-01 03:18:29 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

fix load index for stopping node (#28047)

Browse Source 
Signed-off-by: Wei Liu <wei.liu@zilliz.com>
This commit is contained in:
wei liu 2023-11-03 07:58:18 +08:00 committed by GitHub
parent 1e51255c15
commit 86ec6f4832
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 22 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/querycoordv2/checkers/controller.go
View File

@ -93,7 +93,7 @@ func NewCheckerController(
channelChecker: NewChannelChecker(meta, dist, targetMgr, balancer),
segmentChecker: NewSegmentChecker(meta, dist, targetMgr, balancer, nodeMgr),
balanceChecker: NewBalanceChecker(meta, balancer, nodeMgr, scheduler),
indexChecker: NewIndexChecker(meta, dist, broker),
indexChecker: NewIndexChecker(meta, dist, broker, nodeMgr),
}
manualCheckChs := map[checkerType]chan struct{}{

20
internal/querycoordv2/checkers/index_checker.go
View File

@ -26,6 +26,7 @@ import (
"github.com/milvus-io/milvus/internal/proto/querypb"
"github.com/milvus-io/milvus/internal/querycoordv2/meta"
"github.com/milvus-io/milvus/internal/querycoordv2/params"
"github.com/milvus-io/milvus/internal/querycoordv2/session"
"github.com/milvus-io/milvus/internal/querycoordv2/task"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/typeutil"
@ -35,20 +36,23 @@ var _ Checker = (*IndexChecker)(nil)
// IndexChecker perform segment index check.
type IndexChecker struct {
meta *meta.Meta
dist *meta.DistributionManager
broker meta.Broker
meta *meta.Meta
dist *meta.DistributionManager
broker meta.Broker
nodeMgr *session.NodeManager
}
func NewIndexChecker(
meta *meta.Meta,
dist *meta.DistributionManager,
broker meta.Broker,
nodeMgr *session.NodeManager,
) *IndexChecker {
return &IndexChecker{
meta: meta,
dist: dist,
broker: broker,
meta: meta,
dist: dist,
broker: broker,
nodeMgr: nodeMgr,
}
}
@ -90,6 +94,10 @@ func (c *IndexChecker) checkReplica(ctx context.Context, collection *meta.Collec
targets := make(map[int64][]int64) // segmentID => FieldID
for _, segment := range segments {
// skip update index in stopping node
if ok, _ := c.nodeMgr.IsStoppingNode(segment.Node); ok {
continue
}
missing := c.checkSegment(ctx, segment, collection)
if len(missing) > 0 {
targets[segment.GetID()] = missing

8
internal/querycoordv2/checkers/index_checker_test.go
View File

@ -72,7 +72,7 @@ func (suite *IndexCheckerSuite) SetupTest() {
distManager := meta.NewDistributionManager()
suite.broker = meta.NewMockBroker(suite.T())
suite.checker = NewIndexChecker(suite.meta, distManager, suite.broker)
suite.checker = NewIndexChecker(suite.meta, distManager, suite.broker, suite.nodeMgr)
}
func (suite *IndexCheckerSuite) TearDownTest() {
@ -117,6 +117,12 @@ func (suite *IndexCheckerSuite) TestLoadIndex() {
suite.EqualValues(200, t.ReplicaID())
suite.Equal(task.ActionTypeUpdate, action.Type())
suite.EqualValues(2, action.SegmentID())
// test skip load index for stopping node
suite.nodeMgr.Stopping(1)
suite.nodeMgr.Stopping(2)
tasks = checker.Check(context.Background())
suite.Require().Len(tasks, 0)
}
func (suite *IndexCheckerSuite) TestIndexInfoNotMatch() {