enhance: Don't expire ShardLeaders cache actively (#29879)

issue: #29772 

The shardLeaders cache does not actively expire, update the cache when
search/query fails.

Signed-off-by: Cai Zhang <cai.zhang@zilliz.com>

internal/proxy/meta_cache.go

@@ -23,7 +23,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/samber/lo"
@@ -36,7 +35,6 @@ import (
 	"github.com/milvus-io/milvus/internal/proto/internalpb"
 	"github.com/milvus-io/milvus/internal/proto/querypb"
 	"github.com/milvus-io/milvus/internal/proto/rootcoordpb"
-	"github.com/milvus-io/milvus/internal/querycoordv2/params"
 	"github.com/milvus-io/milvus/internal/types"
 	"github.com/milvus-io/milvus/pkg/common"
 	"github.com/milvus-io/milvus/pkg/log"
@@ -73,7 +71,6 @@ type Cache interface {
 	GetCollectionSchema(ctx context.Context, database, collectionName string) (*schemaInfo, error)
 	GetShards(ctx context.Context, withCache bool, database, collectionName string, collectionID int64) (map[string][]nodeInfo, error)
 	DeprecateShardCache(database, collectionName string)
-	expireShardLeaderCache(ctx context.Context)
 	RemoveCollection(ctx context.Context, database, collectionName string)
 	RemoveCollectionsByID(ctx context.Context, collectionID UniqueID) []string
 	RemovePartition(ctx context.Context, database, collectionName string, partitionName string)
@@ -270,7 +267,6 @@ func InitMetaCache(ctx context.Context, rootCoord types.RootCoordClient, queryCo
 	}
 	globalMetaCache.InitPolicyInfo(resp.PolicyInfos, resp.UserRoles)
 	log.Info("success to init meta cache", zap.Strings("policy_infos", resp.PolicyInfos))
-	globalMetaCache.expireShardLeaderCache(ctx)
 	return nil
 }
 
@@ -875,33 +871,6 @@ func (m *MetaCache) DeprecateShardCache(database, collectionName string) {
 	}
 }
 
-func (m *MetaCache) expireShardLeaderCache(ctx context.Context) {
-	log := log.Ctx(ctx).WithRateGroup("proxy.expireShardLeaderCache", 1, 60)
-	go func() {
-		ticker := time.NewTicker(params.Params.ProxyCfg.ShardLeaderCacheInterval.GetAsDuration(time.Second))
-		defer ticker.Stop()
-
-		for {
-			select {
-			case <-ctx.Done():
-				log.Info("stop periodically update meta cache")
-				return
-			case <-ticker.C:
-				m.leaderMut.RLock()
-				for database, db := range m.collLeader {
-					log.RatedInfo(10, "expire all shard leader cache",
-						zap.String("database", database),
-						zap.Strings("collections", lo.Keys(db)))
-					for _, shards := range db {
-						shards.deprecated.Store(true)
-					}
-				}
-				m.leaderMut.RUnlock()
-			}
-		}
-	}()
-}
-
 func (m *MetaCache) InitPolicyInfo(info []string, userRoles []string) {
 	defer func() {
 		err := getEnforcer().LoadPolicy()

internal/proxy/meta_cache_test.go

@@ -42,7 +42,6 @@ import (
 	"github.com/milvus-io/milvus/pkg/util/crypto"
 	"github.com/milvus-io/milvus/pkg/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/util/merr"
-	"github.com/milvus-io/milvus/pkg/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
 )
 
@@ -762,92 +761,6 @@ func TestMetaCache_RemoveCollection(t *testing.T) {
 	assert.Equal(t, rootCoord.GetAccessCount(), 4)
 }
 
-func TestMetaCache_ExpireShardLeaderCache(t *testing.T) {
-	paramtable.Init()
-	paramtable.Get().Save(Params.ProxyCfg.ShardLeaderCacheInterval.Key, "1")
-
-	ctx := context.Background()
-	rootCoord := &MockRootCoordClientInterface{}
-	queryCoord := &mocks.MockQueryCoordClient{}
-	shardMgr := newShardClientMgr()
-	err := InitMetaCache(ctx, rootCoord, queryCoord, shardMgr)
-	assert.NoError(t, err)
-
-	queryCoord.EXPECT().GetShardLeaders(mock.Anything, mock.Anything).Return(&querypb.GetShardLeadersResponse{
-		Status: merr.Success(),
-		Shards: []*querypb.ShardLeadersList{
-			{
-				ChannelName: "channel-1",
-				NodeIds:     []int64{1, 2, 3},
-				NodeAddrs:   []string{"localhost:9000", "localhost:9001", "localhost:9002"},
-			},
-		},
-	}, nil)
-
-	nodeInfos, err := globalMetaCache.GetShards(ctx, true, dbName, "collection1", 1)
-	assert.NoError(t, err)
-	assert.Len(t, nodeInfos["channel-1"], 3)
-
-	queryCoord.ExpectedCalls = nil
-	queryCoord.EXPECT().GetShardLeaders(mock.Anything, mock.Anything).Return(&querypb.GetShardLeadersResponse{
-		Status: merr.Success(),
-		Shards: []*querypb.ShardLeadersList{
-			{
-				ChannelName: "channel-1",
-				NodeIds:     []int64{1, 2},
-				NodeAddrs:   []string{"localhost:9000", "localhost:9001"},
-			},
-		},
-	}, nil)
-
-	assert.Eventually(t, func() bool {
-		nodeInfos, err := globalMetaCache.GetShards(ctx, true, dbName, "collection1", 1)
-		assert.NoError(t, err)
-		return len(nodeInfos["channel-1"]) == 2
-	}, 3*time.Second, 1*time.Second)
-
-	queryCoord.ExpectedCalls = nil
-	queryCoord.EXPECT().GetShardLeaders(mock.Anything, mock.Anything).Return(&querypb.GetShardLeadersResponse{
-		Status: merr.Success(),
-		Shards: []*querypb.ShardLeadersList{
-			{
-				ChannelName: "channel-1",
-				NodeIds:     []int64{1, 2, 3},
-				NodeAddrs:   []string{"localhost:9000", "localhost:9001", "localhost:9002"},
-			},
-		},
-	}, nil)
-
-	assert.Eventually(t, func() bool {
-		nodeInfos, err := globalMetaCache.GetShards(ctx, true, dbName, "collection1", 1)
-		assert.NoError(t, err)
-		return len(nodeInfos["channel-1"]) == 3
-	}, 3*time.Second, 1*time.Second)
-
-	queryCoord.ExpectedCalls = nil
-	queryCoord.EXPECT().GetShardLeaders(mock.Anything, mock.Anything).Return(&querypb.GetShardLeadersResponse{
-		Status: merr.Success(),
-		Shards: []*querypb.ShardLeadersList{
-			{
-				ChannelName: "channel-1",
-				NodeIds:     []int64{1, 2, 3},
-				NodeAddrs:   []string{"localhost:9000", "localhost:9001", "localhost:9002"},
-			},
-			{
-				ChannelName: "channel-2",
-				NodeIds:     []int64{1, 2, 3},
-				NodeAddrs:   []string{"localhost:9000", "localhost:9001", "localhost:9002"},
-			},
-		},
-	}, nil)
-
-	assert.Eventually(t, func() bool {
-		nodeInfos, err := globalMetaCache.GetShards(ctx, true, dbName, "collection1", 1)
-		assert.NoError(t, err)
-		return len(nodeInfos["channel-1"]) == 3 && len(nodeInfos["channel-2"]) == 3
-	}, 3*time.Second, 1*time.Second)
-}
-
 func TestGlobalMetaCache_ShuffleShardLeaders(t *testing.T) {
 	shards := map[string][]nodeInfo{
 		"channel-1": {

internal/proxy/mock_cache.go

@@ -1039,39 +1039,6 @@ func (_c *MockCache_UpdateCredential_Call) RunAndReturn(run func(*internalpb.Cre
 	return _c
 }
 
-// expireShardLeaderCache provides a mock function with given fields: ctx
-func (_m *MockCache) expireShardLeaderCache(ctx context.Context) {
-	_m.Called(ctx)
-}
-
-// MockCache_expireShardLeaderCache_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'expireShardLeaderCache'
-type MockCache_expireShardLeaderCache_Call struct {
-	*mock.Call
-}
-
-// expireShardLeaderCache is a helper method to define mock.On call
-//   - ctx context.Context
-func (_e *MockCache_Expecter) expireShardLeaderCache(ctx interface{}) *MockCache_expireShardLeaderCache_Call {
-	return &MockCache_expireShardLeaderCache_Call{Call: _e.mock.On("expireShardLeaderCache", ctx)}
-}
-
-func (_c *MockCache_expireShardLeaderCache_Call) Run(run func(ctx context.Context)) *MockCache_expireShardLeaderCache_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context))
-	})
-	return _c
-}
-
-func (_c *MockCache_expireShardLeaderCache_Call) Return() *MockCache_expireShardLeaderCache_Call {
-	_c.Call.Return()
-	return _c
-}
-
-func (_c *MockCache_expireShardLeaderCache_Call) RunAndReturn(run func(context.Context)) *MockCache_expireShardLeaderCache_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
 // NewMockCache creates a new instance of MockCache. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewMockCache(t interface {