forked from kubernetes/autoscaler
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ScaleUp for check-capacity ProvisioningRequestClass (kubernetes#6451)
* ScaleUp for check-capacity ProvisioningRequestClass * update condition logic * Update tests * Naming update * Update cluster-autoscaler/core/scaleup/orchestrator/wrapper_orchestrator_test.go Co-authored-by: Bartek Wróblewski <[email protected]> --------- Co-authored-by: Bartek Wróblewski <[email protected]>
- Loading branch information
1 parent
cf171a7
commit ed6ebbe
Showing
18 changed files
with
1,395 additions
and
50 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
112 changes: 112 additions & 0 deletions
112
cluster-autoscaler/core/scaleup/orchestrator/wrapper_orchestrator.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| /* | ||
| Copyright 2024 The Kubernetes 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 orchestrator | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| appsv1 "k8s.io/api/apps/v1" | ||
| apiv1 "k8s.io/api/core/v1" | ||
| "k8s.io/autoscaler/cluster-autoscaler/clusterstate" | ||
| "k8s.io/autoscaler/cluster-autoscaler/context" | ||
| "k8s.io/autoscaler/cluster-autoscaler/core/scaleup" | ||
| ca_processors "k8s.io/autoscaler/cluster-autoscaler/processors" | ||
| "k8s.io/autoscaler/cluster-autoscaler/processors/provreq" | ||
| "k8s.io/autoscaler/cluster-autoscaler/processors/status" | ||
| "k8s.io/autoscaler/cluster-autoscaler/provisioningrequest/checkcapacity" | ||
| "k8s.io/autoscaler/cluster-autoscaler/utils/errors" | ||
| "k8s.io/autoscaler/cluster-autoscaler/utils/taints" | ||
| "k8s.io/client-go/rest" | ||
| schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework" | ||
| ) | ||
|
|
||
| // WrapperOrchestrator is an orchestrator which wraps Scale Up for ProvisioningRequests and regular pods. | ||
| // Each loop WrapperOrchestrator split out regular and pods from ProvisioningRequest, pick one group that | ||
| // wasn't picked in the last loop and run ScaleUp for it. | ||
| type WrapperOrchestrator struct { | ||
| // scaleUpRegularPods indicates that ScaleUp for regular pods will be run in the current CA loop, if they are present. | ||
| scaleUpRegularPods bool | ||
| scaleUpOrchestrator scaleup.Orchestrator | ||
| provReqOrchestrator scaleup.Orchestrator | ||
| } | ||
|
|
||
| // NewWrapperOrchestrator return WrapperOrchestrator | ||
| func NewWrapperOrchestrator(kubeConfig *rest.Config) (scaleup.Orchestrator, error) { | ||
| provReqOrchestrator, err := checkcapacity.New(kubeConfig) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed create ScaleUp orchestrator for ProvisioningRequests, error: %v", err) | ||
| } | ||
| return &WrapperOrchestrator{ | ||
| scaleUpOrchestrator: New(), | ||
| provReqOrchestrator: provReqOrchestrator, | ||
| }, nil | ||
| } | ||
|
|
||
| // Initialize initializes the orchestrator object with required fields. | ||
| func (o *WrapperOrchestrator) Initialize( | ||
| autoscalingContext *context.AutoscalingContext, | ||
| processors *ca_processors.AutoscalingProcessors, | ||
| clusterStateRegistry *clusterstate.ClusterStateRegistry, | ||
| taintConfig taints.TaintConfig, | ||
| ) { | ||
| o.scaleUpOrchestrator.Initialize(autoscalingContext, processors, clusterStateRegistry, taintConfig) | ||
| o.provReqOrchestrator.Initialize(autoscalingContext, processors, clusterStateRegistry, taintConfig) | ||
| } | ||
|
|
||
| // ScaleUp run scaleUp function for regular pods of pods from ProvisioningRequest. | ||
| func (o *WrapperOrchestrator) ScaleUp( | ||
| unschedulablePods []*apiv1.Pod, | ||
| nodes []*apiv1.Node, | ||
| daemonSets []*appsv1.DaemonSet, | ||
| nodeInfos map[string]*schedulerframework.NodeInfo, | ||
| ) (*status.ScaleUpStatus, errors.AutoscalerError) { | ||
| defer func() { o.scaleUpRegularPods = !o.scaleUpRegularPods }() | ||
|
|
||
| provReqPods, regularPods := splitOut(unschedulablePods) | ||
| if len(provReqPods) == 0 { | ||
| o.scaleUpRegularPods = true | ||
| } else if len(regularPods) == 0 { | ||
| o.scaleUpRegularPods = false | ||
| } | ||
|
|
||
| if o.scaleUpRegularPods { | ||
| return o.scaleUpOrchestrator.ScaleUp(regularPods, nodes, daemonSets, nodeInfos) | ||
| } | ||
| return o.provReqOrchestrator.ScaleUp(provReqPods, nodes, daemonSets, nodeInfos) | ||
| } | ||
|
|
||
| func splitOut(unschedulablePods []*apiv1.Pod) (provReqPods, regularPods []*apiv1.Pod) { | ||
| for _, pod := range unschedulablePods { | ||
| if _, ok := pod.Annotations[provreq.ProvisioningRequestPodAnnotationKey]; ok { | ||
| provReqPods = append(provReqPods, pod) | ||
| } else { | ||
| regularPods = append(regularPods, pod) | ||
| } | ||
| } | ||
| return | ||
| } | ||
|
|
||
| // ScaleUpToNodeGroupMinSize tries to scale up node groups that have less nodes | ||
| // than the configured min size. The source of truth for the current node group | ||
| // size is the TargetSize queried directly from cloud providers. Returns | ||
| // appropriate status or error if an unexpected error occurred. | ||
| func (o *WrapperOrchestrator) ScaleUpToNodeGroupMinSize( | ||
| nodes []*apiv1.Node, | ||
| nodeInfos map[string]*schedulerframework.NodeInfo, | ||
| ) (*status.ScaleUpStatus, errors.AutoscalerError) { | ||
| return o.scaleUpOrchestrator.ScaleUpToNodeGroupMinSize(nodes, nodeInfos) | ||
| } |
90 changes: 90 additions & 0 deletions
90
cluster-autoscaler/core/scaleup/orchestrator/wrapper_orchestrator_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| /* | ||
| Copyright 2024 The Kubernetes 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 orchestrator | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| appsv1 "k8s.io/api/apps/v1" | ||
| apiv1 "k8s.io/api/core/v1" | ||
| "k8s.io/autoscaler/cluster-autoscaler/clusterstate" | ||
| "k8s.io/autoscaler/cluster-autoscaler/context" | ||
| ca_processors "k8s.io/autoscaler/cluster-autoscaler/processors" | ||
| "k8s.io/autoscaler/cluster-autoscaler/processors/provreq" | ||
| "k8s.io/autoscaler/cluster-autoscaler/processors/status" | ||
| "k8s.io/autoscaler/cluster-autoscaler/utils/errors" | ||
| "k8s.io/autoscaler/cluster-autoscaler/utils/taints" | ||
| . "k8s.io/autoscaler/cluster-autoscaler/utils/test" | ||
| schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework" | ||
| ) | ||
|
|
||
| const ( | ||
| provisioningRequestErrorMsg = "provisioningRequestError" | ||
| regularPodsErrorMsg = "regularPodsError" | ||
| ) | ||
|
|
||
| func TestScaleUp(t *testing.T) { | ||
| o := WrapperOrchestrator{ | ||
| provReqOrchestrator: &fakeScaleUp{provisioningRequestErrorMsg}, | ||
| scaleUpOrchestrator: &fakeScaleUp{regularPodsErrorMsg}, | ||
| } | ||
| regularPods := []*apiv1.Pod{ | ||
| BuildTestPod("pod-1", 1, 100), | ||
| BuildTestPod("pod-2", 1, 100), | ||
| } | ||
| provReqPods := []*apiv1.Pod{ | ||
| BuildTestPod("pr-pod-1", 1, 100), | ||
| BuildTestPod("pr-pod-2", 1, 100), | ||
| } | ||
| for _, pod := range provReqPods { | ||
| pod.Annotations[provreq.ProvisioningRequestPodAnnotationKey] = "true" | ||
| } | ||
| unschedulablePods := append(regularPods, provReqPods...) | ||
| _, err := o.ScaleUp(unschedulablePods, nil, nil, nil) | ||
| assert.Equal(t, err.Error(), provisioningRequestErrorMsg) | ||
| _, err = o.ScaleUp(unschedulablePods, nil, nil, nil) | ||
| assert.Equal(t, err.Error(), regularPodsErrorMsg) | ||
| } | ||
|
|
||
| type fakeScaleUp struct { | ||
| errorMsg string | ||
| } | ||
|
|
||
| func (f *fakeScaleUp) ScaleUp( | ||
| unschedulablePods []*apiv1.Pod, | ||
| nodes []*apiv1.Node, | ||
| daemonSets []*appsv1.DaemonSet, | ||
| nodeInfos map[string]*schedulerframework.NodeInfo, | ||
| ) (*status.ScaleUpStatus, errors.AutoscalerError) { | ||
| return nil, errors.NewAutoscalerError(errors.InternalError, f.errorMsg) | ||
| } | ||
|
|
||
| func (f *fakeScaleUp) Initialize( | ||
| autoscalingContext *context.AutoscalingContext, | ||
| processors *ca_processors.AutoscalingProcessors, | ||
| clusterStateRegistry *clusterstate.ClusterStateRegistry, | ||
| taintConfig taints.TaintConfig, | ||
| ) { | ||
| } | ||
|
|
||
| func (f *fakeScaleUp) ScaleUpToNodeGroupMinSize( | ||
| nodes []*apiv1.Node, | ||
| nodeInfos map[string]*schedulerframework.NodeInfo, | ||
| ) (*status.ScaleUpStatus, errors.AutoscalerError) { | ||
| return nil, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.