scheduler: tolerate partial cluster failures in replica estimation

pkg/estimator/client/accurate.go

@@ -267,6 +267,9 @@ func getClusterReplicasConcurrently(parentCtx context.Context, clusters []string
 		funcs[index] = func() error {
 			replicas, err := getClusterReplicas(ctx, localCluster)
 			if err != nil {
+				// Mark the failed cluster instead of dropping the whole batch, so one
+				// cluster's failure does not discard the replicas estimated for the others.
+				clusterReplicas[localIndex] = workv1alpha2.TargetCluster{Name: localCluster, Replicas: UnauthenticReplica}
 				return err
 			}
 			clusterReplicas[localIndex] = workv1alpha2.TargetCluster{Name: localCluster, Replicas: replicas}
@@ -296,6 +299,9 @@ func getClusterComponentSetsConcurrently(
 		funcs[i] = func() error {
 			sets, err := getClusterSetsEstimation(ctx, localCluster.Name)
 			if err != nil {
+				// Mark the failed cluster instead of dropping the whole batch, so one
+				// cluster's failure does not discard the sets estimated for the others.
+				results[localIndex] = ComponentSetEstimationResponse{Name: localCluster.Name, Sets: UnauthenticReplica}
 				return err
 			}
 			results[localIndex] = ComponentSetEstimationResponse{Name: localCluster.Name, Sets: sets}

pkg/estimator/client/accurate_test.go

@@ -18,6 +18,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -162,6 +163,64 @@ func Test_toAssumedWorkload(t *testing.T) {
 	}
 }
 
+func Test_getClusterReplicasConcurrently_partialFailure(t *testing.T) {
+	clusters := []string{"cluster1", "cluster2", "cluster3"}
+
+	tests := []struct {
+		name         string
+		getReplicas  getClusterReplicasFunc
+		wantReplicas map[string]int32
+		wantErr      bool
+	}{
+		{
+			name: "all clusters succeed",
+			getReplicas: func(_ context.Context, cluster string) (int32, error) {
+				return map[string]int32{"cluster1": 10, "cluster2": 20, "cluster3": 30}[cluster], nil
+			},
+			wantReplicas: map[string]int32{"cluster1": 10, "cluster2": 20, "cluster3": 30},
+			wantErr:      false,
+		},
+		{
+			name: "one cluster fails — others keep their replicas, failed one is UnauthenticReplica",
+			getReplicas: func(_ context.Context, cluster string) (int32, error) {
+				if cluster == "cluster2" {
+					return 0, errors.New("estimator unavailable")
+				}
+				return map[string]int32{"cluster1": 10, "cluster3": 30}[cluster], nil
+			},
+			wantReplicas: map[string]int32{"cluster1": 10, "cluster2": UnauthenticReplica, "cluster3": 30},
+			wantErr:      true,
+		},
+		{
+			name: "all clusters fail — every cluster marked UnauthenticReplica with error",
+			getReplicas: func(_ context.Context, _ string) (int32, error) {
+				return 0, errors.New("estimator unavailable")
+			},
+			wantReplicas: map[string]int32{"cluster1": UnauthenticReplica, "cluster2": UnauthenticReplica, "cluster3": UnauthenticReplica},
+			wantErr:      true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := getClusterReplicasConcurrently(context.Background(), clusters, 5*time.Second, tt.getReplicas)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+
+			// Every cluster keeps a named slot in the result, even on failure.
+			require.Len(t, got, len(clusters))
+			for _, tc := range got {
+				want, ok := tt.wantReplicas[tc.Name]
+				require.True(t, ok, "unexpected cluster %q in result", tc.Name)
+				assert.Equal(t, want, tc.Replicas, "cluster %q replicas", tc.Name)
+			}
+		})
+	}
+}
+
 func Test_maxAvailableReplicas_assumedWorkloads(t *testing.T) {
 	const clusterName = "cluster-a"
 	const wantMaxReplicas = int32(10)

pkg/scheduler/core/estimation.go

@@ -85,9 +85,10 @@ func calculateMultiTemplateAvailableSets(ctx context.Context, estCtx multiTempla
 	namespacedKey := names.NamespacedKey(estCtx.spec.Resource.Namespace, estCtx.spec.Resource.Name)
 	resp, err := estCtx.estimator.MaxAvailableComponentSets(ctx, req)
 	if err != nil {
+		// Don't return early: on a partial failure resp still carries the clusters that
+		// succeeded, so fall through and build a result from what is available.
 		klog.Errorf("Failed to calculate available component set with estimator(%s) for workload(%s, kind=%s, %s): %v",
 			estCtx.estimatorName, estCtx.spec.Resource.APIVersion, estCtx.spec.Resource.Kind, namespacedKey, err)
-		return nil, err
 	}
 
 	// Use a map to safely update replicas regardless of order.
@@ -109,7 +110,7 @@ func calculateMultiTemplateAvailableSets(ctx context.Context, estCtx multiTempla
 		}
 		result = append(result, workv1alpha2.TargetCluster{Name: cluster.Name, Replicas: sets})
 	}
-	return result, nil
+	return result, err
 }
 
 // buildAssumedWorkloadsByCluster builds a map of assumed workloads for each cluster based on the assigning cache.

pkg/scheduler/core/estimation_test.go

@@ -33,13 +33,15 @@ import (
 
 // mockReplicaEstimator is a mock implementation of ReplicaEstimator for testing
 type mockReplicaEstimator struct {
+	maxAvailableReplicasResponse      []workv1alpha2.TargetCluster
+	maxAvailableReplicasError         error
 	maxAvailableComponentSetsResponse []estimatorclient.ComponentSetEstimationResponse
 	maxAvailableComponentSetsError    error
 	lastComponentSetRequest           estimatorclient.ComponentSetEstimationRequest
 }
 
 func (m *mockReplicaEstimator) MaxAvailableReplicas(_ context.Context, _ estimatorclient.ReplicaEstimationRequest) ([]workv1alpha2.TargetCluster, error) {
-	return nil, nil
+	return m.maxAvailableReplicasResponse, m.maxAvailableReplicasError
 }
 
 func (m *mockReplicaEstimator) MaxAvailableComponentSets(_ context.Context, req estimatorclient.ComponentSetEstimationRequest) ([]estimatorclient.ComponentSetEstimationResponse, error) {
@@ -290,12 +292,26 @@ func Test_calculateMultiTemplateAvailableSets(t *testing.T) {
 			},
 		},
 		{
-			name:           "estimator error — returns nil result",
+			name:           "estimator error with no results — returns empty result and propagates error",
 			mockResponse:   nil,
 			mockError:      errors.New("estimator error"),
-			expectedResult: nil,
+			expectedResult: []workv1alpha2.TargetCluster{},
 			expectedError:  true,
 		},
+		{
+			name: "partial failure — successful clusters returned, failed cluster skipped, error propagated",
+			mockResponse: []estimatorclient.ComponentSetEstimationResponse{
+				{Name: "cluster1", Sets: 50},
+				{Name: "cluster2", Sets: estimatorclient.UnauthenticReplica},
+				{Name: "cluster3", Sets: 250},
+			},
+			mockError: errors.New("cluster2: estimator unavailable"),
+			expectedResult: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 50},
+				{Name: "cluster3", Replicas: 250},
+			},
+			expectedError: true,
+		},
 		{
 			name:           "empty estimator response — returns empty result",
 			mockResponse:   []estimatorclient.ComponentSetEstimationResponse{},
@@ -329,6 +345,100 @@ func Test_calculateMultiTemplateAvailableSets(t *testing.T) {
 	}
 }
 
+func Test_runSingleTemplateEstimator_partialFailure(t *testing.T) {
+	clusters := []*clusterv1alpha1.Cluster{
+		helper.NewCluster("cluster1"),
+		helper.NewCluster("cluster2"),
+	}
+	spec := &workv1alpha2.ResourceBindingSpec{
+		Resource: workv1alpha2.ObjectReference{
+			APIVersion: "apps/v1",
+			Kind:       "Deployment",
+			Namespace:  "default",
+			Name:       "test-deployment",
+		},
+		Replicas: 1,
+	}
+
+	// cluster2 failed (UnauthenticReplica) and an error is returned; the path must still
+	// surface the partial result rather than discarding it.
+	partial := []workv1alpha2.TargetCluster{
+		{Name: "cluster1", Replicas: 5},
+		{Name: "cluster2", Replicas: estimatorclient.UnauthenticReplica},
+	}
+	mockEstimator := &mockReplicaEstimator{
+		maxAvailableReplicasResponse: partial,
+		maxAvailableReplicasError:    errors.New("cluster2: estimator unavailable"),
+	}
+
+	res, err := runSingleTemplateEstimator(context.Background(), replicaEstimationContext{
+		estimatorName: "test-estimator",
+		estimator:     mockEstimator,
+		clusters:      clusters,
+		spec:          spec,
+	})
+
+	assert.Error(t, err)
+	assert.Equal(t, partial, res)
+}
+
+func Test_mergeReplicaResults(t *testing.T) {
+	tests := []struct {
+		name      string
+		available []workv1alpha2.TargetCluster
+		res       []workv1alpha2.TargetCluster
+		want      []workv1alpha2.TargetCluster
+	}{
+		{
+			name: "takes the minimum replicas per cluster",
+			available: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 10},
+				{Name: "cluster2", Replicas: 10},
+			},
+			res: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 4},
+				{Name: "cluster2", Replicas: 20},
+			},
+			want: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 4},
+				{Name: "cluster2", Replicas: 10},
+			},
+		},
+		{
+			name: "skips clusters marked UnauthenticReplica so a partial failure does not corrupt them",
+			available: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 10},
+				{Name: "cluster2", Replicas: 10},
+			},
+			res: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 4},
+				{Name: "cluster2", Replicas: estimatorclient.UnauthenticReplica},
+			},
+			want: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 4},
+				{Name: "cluster2", Replicas: 10},
+			},
+		},
+		{
+			name: "nil result leaves available unchanged",
+			available: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 10},
+			},
+			res: nil,
+			want: []workv1alpha2.TargetCluster{
+				{Name: "cluster1", Replicas: 10},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := mergeReplicaResults(tt.available, tt.res)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
 func Test_buildAssumedWorkloadsByCluster(t *testing.T) {
 	t.Run("nil cache returns empty map", func(t *testing.T) {
 		clusters := []*clusterv1alpha1.Cluster{

pkg/scheduler/core/util.go

@@ -92,7 +92,10 @@ func calAvailableReplicas(clusters []*clusterv1alpha1.Cluster, spec *workv1alpha
 			assumedWorkloads: assumedWorkloads,
 		})
 		if err != nil {
-			continue
+			// On a partial failure res still holds the clusters that succeeded; failed ones
+			// are marked UnauthenticReplica and skipped by mergeReplicaResults. Keep merging
+			// instead of discarding the estimator for every cluster.
+			klog.Errorf("Estimator %s returned an error, merging any partial results: %v", name, err)
 		}
 		availableTargetClusters = mergeReplicaResults(availableTargetClusters, res)
 	}
@@ -145,8 +148,10 @@ func runSingleTemplateEstimator(ctx context.Context, in replicaEstimationContext
 		AssumedWorkloads:    in.assumedWorkloads,
 	})
 	if err != nil {
+		// Return res alongside the error so the caller can still use the clusters that
+		// succeeded; failed clusters are marked UnauthenticReplica.
 		klog.Errorf("Max cluster available replicas error: %v", err)
-		return nil, err
+		return res, err
 	}
 	klog.V(4).Infof("Invoked MaxAvailableReplicas of estimator %s for workload(%s, kind=%s, %s): %v", in.estimatorName,
 		in.spec.Resource.APIVersion, in.spec.Resource.Kind, names.NamespacedKey(in.spec.Resource.Namespace, in.spec.Resource.Name), res)