From 7254d8b45418e1bece94b1760b1bf1e1813958cb Mon Sep 17 00:00:00 2001 From: aeron-gh Date: Thu, 25 Jun 2026 00:25:59 +0530 Subject: [PATCH] scheduler: tolerate partial cluster failures in replica estimation When the scheduler-estimator failed to estimate replicas for a single cluster, the entire estimator result was discarded and every cluster fell back to the general-estimator, even the clusters that had been estimated successfully. Mark a failed cluster as UnauthenticReplica instead of dropping the whole batch, and let the caller merge the partial results. mergeReplicaResults already skips UnauthenticReplica entries, so only the clusters that succeeded are merged while a failed cluster keeps its previous value. The same handling is applied to the multi-template component-set path. The descheduler path keeps its existing all-or-nothing behavior and is unaffected. Signed-off-by: aeron-gh --- pkg/estimator/client/accurate.go | 6 ++ pkg/estimator/client/accurate_test.go | 59 +++++++++++++ pkg/scheduler/core/estimation.go | 5 +- pkg/scheduler/core/estimation_test.go | 116 +++++++++++++++++++++++++- pkg/scheduler/core/util.go | 9 +- 5 files changed, 188 insertions(+), 7 deletions(-) diff --git a/pkg/estimator/client/accurate.go b/pkg/estimator/client/accurate.go index b2ec49dc9754..f68df5f2e67d 100644 --- a/pkg/estimator/client/accurate.go +++ b/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} diff --git a/pkg/estimator/client/accurate_test.go b/pkg/estimator/client/accurate_test.go index 247fdada6fd8..8d0791be43fa 100644 --- a/pkg/estimator/client/accurate_test.go +++ b/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) diff --git a/pkg/scheduler/core/estimation.go b/pkg/scheduler/core/estimation.go index 2b012438a093..b96994b3165c 100644 --- a/pkg/scheduler/core/estimation.go +++ b/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. diff --git a/pkg/scheduler/core/estimation_test.go b/pkg/scheduler/core/estimation_test.go index ff2b75f640b9..33edf313e706 100644 --- a/pkg/scheduler/core/estimation_test.go +++ b/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{ diff --git a/pkg/scheduler/core/util.go b/pkg/scheduler/core/util.go index 1495af01ac22..1bcf358caa70 100644 --- a/pkg/scheduler/core/util.go +++ b/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)