admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-08 10:08:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: Make sure querycoord observers started once (#35811)

Browse Source 
Related to #35809

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2024-08-29 14:45:00 +08:00 committed by GitHub
parent 70bc0e4538
commit 09ef3f1b4f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 56 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -56,7 +56,8 @@ type CollectionObserver struct {
proxyManager proxyutil.ProxyClientManagerInterface proxyManager proxyutil.ProxyClientManagerInterface
stopOnce sync.Once startOnce sync.Once
stopOnce sync.Once
} }
type LoadTask struct { type LoadTask struct {
@ -94,27 +95,29 @@ func NewCollectionObserver(
} }
func (ob *CollectionObserver) Start() { func (ob *CollectionObserver) Start() {
ctx, cancel := context.WithCancel(context.Background()) ob.startOnce.Do(func() {
ob.cancel = cancel ctx, cancel := context.WithCancel(context.Background())
ob.cancel = cancel
observePeriod := Params.QueryCoordCfg.CollectionObserverInterval.GetAsDuration(time.Millisecond) observePeriod := Params.QueryCoordCfg.CollectionObserverInterval.GetAsDuration(time.Millisecond)
ob.wg.Add(1) ob.wg.Add(1)
go func() { go func() {
defer ob.wg.Done() defer ob.wg.Done()
ticker := time.NewTicker(observePeriod) ticker := time.NewTicker(observePeriod)
defer ticker.Stop() defer ticker.Stop()
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
log.Info("CollectionObserver stopped") log.Info("CollectionObserver stopped")
return return
case <-ticker.C: case <-ticker.C:
ob.Observe(ctx) ob.Observe(ctx)
}
} }
} }()
}() })
} }
func (ob *CollectionObserver) Stop() { func (ob *CollectionObserver) Stop() {

7
internal/querycoordv2/observers/leader_cache_observer.go
View File

@ -36,6 +36,7 @@ type CollectionShardLeaderCache = map[string]*querypb.ShardLeadersList
type LeaderCacheObserver struct { type LeaderCacheObserver struct {
wg sync.WaitGroup wg sync.WaitGroup
proxyManager proxyutil.ProxyClientManagerInterface proxyManager proxyutil.ProxyClientManagerInterface
startOnce sync.Once
stopOnce sync.Once stopOnce sync.Once
closeCh chan struct{} closeCh chan struct{}
@ -44,8 +45,10 @@ type LeaderCacheObserver struct {
} }
func (o *LeaderCacheObserver) Start(ctx context.Context) { func (o *LeaderCacheObserver) Start(ctx context.Context) {
o.wg.Add(1) o.startOnce.Do(func() {
go o.schedule(ctx) o.wg.Add(1)
go o.schedule(ctx)
})
} }
func (o *LeaderCacheObserver) Stop() { func (o *LeaderCacheObserver) Stop() {

13
internal/querycoordv2/observers/replica_observer.go
View File

@ -37,7 +37,8 @@ type ReplicaObserver struct {
meta *meta.Meta meta *meta.Meta
distMgr *meta.DistributionManager distMgr *meta.DistributionManager
stopOnce sync.Once startOnce sync.Once
stopOnce sync.Once
} }
func NewReplicaObserver(meta *meta.Meta, distMgr *meta.DistributionManager) *ReplicaObserver { func NewReplicaObserver(meta *meta.Meta, distMgr *meta.DistributionManager) *ReplicaObserver {
@ -48,11 +49,13 @@ func NewReplicaObserver(meta *meta.Meta, distMgr *meta.DistributionManager) *Rep
} }
func (ob *ReplicaObserver) Start() { func (ob *ReplicaObserver) Start() {
ctx, cancel := context.WithCancel(context.Background()) ob.startOnce.Do(func() {
ob.cancel = cancel ctx, cancel := context.WithCancel(context.Background())
ob.cancel = cancel
ob.wg.Add(1) ob.wg.Add(1)
go ob.schedule(ctx) go ob.schedule(ctx)
})
} }
func (ob *ReplicaObserver) Stop() { func (ob *ReplicaObserver) Stop() {

13
internal/querycoordv2/observers/resource_observer.go
View File

@ -36,7 +36,8 @@ type ResourceObserver struct {
wg sync.WaitGroup wg sync.WaitGroup
meta *meta.Meta meta *meta.Meta
stopOnce sync.Once startOnce sync.Once
stopOnce sync.Once
} }
func NewResourceObserver(meta *meta.Meta) *ResourceObserver { func NewResourceObserver(meta *meta.Meta) *ResourceObserver {
@ -46,11 +47,13 @@ func NewResourceObserver(meta *meta.Meta) *ResourceObserver {
} }
func (ob *ResourceObserver) Start() { func (ob *ResourceObserver) Start() {
ctx, cancel := context.WithCancel(context.Background()) ob.startOnce.Do(func() {
ob.cancel = cancel ctx, cancel := context.WithCancel(context.Background())
ob.cancel = cancel
ob.wg.Add(1) ob.wg.Add(1)
go ob.schedule(ctx) go ob.schedule(ctx)
})
} }
func (ob *ResourceObserver) Stop() { func (ob *ResourceObserver) Stop() {

25
internal/querycoordv2/observers/target_observer.go
View File

@ -71,7 +71,8 @@ type TargetObserver struct {
dispatcher *taskDispatcher[int64] dispatcher *taskDispatcher[int64]
keylocks *lock.KeyLock[int64] keylocks *lock.KeyLock[int64]
stopOnce sync.Once startOnce sync.Once
stopOnce sync.Once
} }
func NewTargetObserver( func NewTargetObserver(
@ -101,19 +102,21 @@ func NewTargetObserver(
} }
func (ob *TargetObserver) Start() { func (ob *TargetObserver) Start() {
ctx, cancel := context.WithCancel(context.Background()) ob.startOnce.Do(func() {
ob.cancel = cancel ctx, cancel := context.WithCancel(context.Background())
ob.cancel = cancel
ob.dispatcher.Start() ob.dispatcher.Start()
ob.wg.Add(1) ob.wg.Add(1)
go func() { go func() {
defer ob.wg.Done() defer ob.wg.Done()
ob.schedule(ctx) ob.schedule(ctx)
}() }()
// after target observer start, update target for all collection // after target observer start, update target for all collection
ob.initChan <- initRequest{} ob.initChan <- initRequest{}
})
} }
func (ob *TargetObserver) Stop() { func (ob *TargetObserver) Stop() {

2
internal/querycoordv2/services_test.go
View File

@ -1084,8 +1084,6 @@ func (suite *ServiceSuite) TestReleasePartition() {
func (suite *ServiceSuite) TestRefreshCollection() { func (suite *ServiceSuite) TestRefreshCollection() {
server := suite.server server := suite.server
server.collectionObserver.Start()
// Test refresh all collections. // Test refresh all collections.
for _, collection := range suite.collections { for _, collection := range suite.collections {
err := server.refreshCollection(collection) err := server.refreshCollection(collection)