milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Only log if the refresh actually happens (#20505) 

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

Signed-off-by: yah01 <yang.cen@zilliz.com>
pull/20507/head
yah01 2022-11-11 14:37:05 +08:00 committed by GitHub
parent faa2df3aac
commit 8f13fcd612
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 19 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -117,10 +117,11 @@ func (ob *CollectionObserver) observeTimeout() {
ob.meta.ReplicaManager.RemoveCollection(collection.GetCollectionID())
ob.targetMgr.RemoveCollection(collection.GetCollectionID())
} else if now.After(refreshTime) {
ob.refreshTargets(collection.UpdatedAt, collection.GetCollectionID())
log.Info("load for long time, refresh targets of collection",
zap.Duration("loadTime", time.Since(collection.CreatedAt)),
)
if ob.refreshTargets(collection.UpdatedAt, collection.GetCollectionID()) {
log.Info("load for long time, refresh targets of collection",
zap.Duration("loadTime", time.Since(collection.CreatedAt)),
)
}
}
}
@ -155,20 +156,25 @@ func (ob *CollectionObserver) observeTimeout() {
partitionIDs := lo.Map(partitions, func(partition *meta.Partition, _ int) int64 {
return partition.GetPartitionID()
})
ob.refreshTargets(partition.UpdatedAt, partition.GetCollectionID(), partitionIDs...)
log.Info("load for long time, refresh targets of partitions",
zap.Duration("loadTime", time.Since(partition.CreatedAt)),
)
if ob.refreshTargets(partition.UpdatedAt, partition.GetCollectionID(), partitionIDs...) {
log.Info("load for long time, refresh targets of partitions",
zap.Duration("loadTime", time.Since(partition.CreatedAt)),
)
}
break
}
}
}
}
func (ob *CollectionObserver) refreshTargets(updatedAt time.Time, collectionID int64, partitions ...int64) {
// refreshTargets refreshes the targets of the given collection,
// avoids repeated refreshing by checking the updatedAt,
// returns true if actually refreshed the targets,
// false otherwise
func (ob *CollectionObserver) refreshTargets(updatedAt time.Time, collectionID int64, partitions ...int64) bool {
refreshedTime, ok := ob.refreshed[collectionID]
if ok && refreshedTime.Equal(updatedAt) {
return
return false
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
@ -181,8 +187,8 @@ func (ob *CollectionObserver) refreshTargets(updatedAt time.Time, collectionID i
var err error
partitions, err = ob.broker.GetPartitions(ctx, collectionID)
if err != nil {
log.Error("failed to get partitions from RootCoord", zap.Error(err))
return
log.Warn("failed to get partitions from RootCoord, will refresh targets later", zap.Error(err))
return false
}
}
@ -195,6 +201,7 @@ func (ob *CollectionObserver) refreshTargets(updatedAt time.Time, collectionID i
)
ob.refreshed[collectionID] = updatedAt
return true
}
func (ob *CollectionObserver) observeLoadStatus() {

29
internal/util/indexparams/disk_index_params_test.go
View File

@ -27,35 +27,6 @@ import (
)
func TestDiskIndexParams(t *testing.T) {
t.Run("fill index params without auto index param", func(t *testing.T) {
var params paramtable.ComponentParam
params.Init()
indexParams := make(map[string]string)
err := FillDiskIndexParams(&params, indexParams)
assert.NoError(t, err)
pgCodeBudgetGBRatio, err := strconv.ParseFloat(indexParams[PQCodeBudgetRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 0.125, pgCodeBudgetGBRatio)
buildNumThreadsRatio, err := strconv.ParseFloat(indexParams[NumBuildThreadRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 1.0, buildNumThreadsRatio)
searchCacheBudgetGBRatio, err := strconv.ParseFloat(indexParams[SearchCacheBudgetRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 0.125, searchCacheBudgetGBRatio)
loadNumThreadRatio, err := strconv.ParseFloat(indexParams[NumLoadThreadRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 8.0, loadNumThreadRatio)
beamWidthRatio, err := strconv.ParseFloat(indexParams[BeamWidthRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 4.0, beamWidthRatio)
})
t.Run("fill index params with auto index", func(t *testing.T) {
var params paramtable.ComponentParam
params.AutoIndexConfig.Enable = true