fix: update scheduler estimator cache when cluster changed

Signed-off-by: zhy76 <958474674@qq.com>

pkg/descheduler/descheduler_test.go

@@ -505,7 +505,7 @@ func TestDescheduler_worker(t *testing.T) {
 					mock.MatchedBy(func(context.Context) bool { return true }),
 					mock.MatchedBy(func(in *pb.UnschedulableReplicasRequest) bool { return in.Cluster == cluster.Name }),
 				).Return(mockResultFn, nil)
-				desched.schedulerEstimatorCache.AddCluster(cluster.Name, nil, mockClient)
+				desched.schedulerEstimatorCache.AddOrUpdateCluster(cluster.Name, nil, mockClient)
 			}
 
 			desched.informerFactory.Start(ctx.Done())

pkg/estimator/client/cache.go

@@ -1,19 +1,3 @@
-/*
-Copyright 2021 The Karmada Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
 package client
 
 import (
@@ -47,6 +31,7 @@ func NewSchedulerEstimatorCache() *SchedulerEstimatorCache {
 type clientWrapper struct {
 	connection *grpc.ClientConn
 	client     estimatorservice.EstimatorClient
+	closeLock  sync.Mutex
 }
 
 // IsEstimatorExist checks whether the cluster estimator exists in the cache.
@@ -57,16 +42,10 @@ func (c *SchedulerEstimatorCache) IsEstimatorExist(name string) bool {
 	return exist
 }
 
-// AddCluster adds a grpc connection and associated client into the cache.
-func (c *SchedulerEstimatorCache) AddCluster(name string, connection *grpc.ClientConn, client estimatorservice.EstimatorClient) {
+// AddOrUpdateCluster adds or updates a grpc connection and associated client into the cache.
+func (c *SchedulerEstimatorCache) AddOrUpdateCluster(name string, connection *grpc.ClientConn, client estimatorservice.EstimatorClient) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
-	// If more than one worker have established connections at the same time,
-	// only the first one will be used. The ladder ones would be abandoned.
-	if _, exist := c.estimator[name]; exist {
-		_ = connection.Close()
-		return
-	}
 	c.estimator[name] = &clientWrapper{
 		connection: connection,
 		client:     client,
@@ -81,6 +60,10 @@ func (c *SchedulerEstimatorCache) DeleteCluster(name string) {
 	if !exist {
 		return
 	}
+	// Use a lock to prevent if more than one worker close connection at the same time,
+	// a connection may be closed multiple times.
+	es.closeLock.Lock()
+	defer es.closeLock.Unlock()
 	_ = es.connection.Close()
 	delete(c.estimator, name)
 }
@@ -98,9 +81,20 @@ func (c *SchedulerEstimatorCache) GetClient(name string) (estimatorservice.Estim
 
 // EstablishConnection establishes a new gRPC connection with the specified cluster scheduler estimator.
 func EstablishConnection(kubeClient kubernetes.Interface, name string, estimatorCache *SchedulerEstimatorCache, estimatorServicePrefix string, grpcConfig *grpcconnection.ClientConfig) error {
-	if estimatorCache.IsEstimatorExist(name) {
-		return nil
+	// check if the connection already exists, if so, close it and establish a new connection
+	estimatorCache.lock.Lock()
+	es, exist := estimatorCache.estimator[name]
+	if exist {
+		es.closeLock.Lock()
+		err := es.connection.Close()
+		es.closeLock.Unlock()
+		if err != nil {
+			estimatorCache.lock.Unlock()
+			klog.Errorf("Failed to close connection of cluster(%s): %v.", name, err)
+			return err
+		}
 	}
+	estimatorCache.lock.Unlock()
 
 	serverAddr, err := resolveCluster(kubeClient, util.NamespaceKarmadaSystem,
 		names.GenerateEstimatorServiceName(estimatorServicePrefix, name), int32(grpcConfig.TargetPort))
@@ -115,7 +109,8 @@ func EstablishConnection(kubeClient kubernetes.Interface, name string, estimator
 		return err
 	}
 	c := estimatorservice.NewEstimatorClient(cc)
-	estimatorCache.AddCluster(name, cc, c)
+	// add or update the cache
+	estimatorCache.AddOrUpdateCluster(name, cc, c)
 	klog.Infof("Connection with estimator server(%s) of cluster(%s) has been established.", serverAddr, name)
 	return nil
 }