Skip to content

Commit

Permalink
feat: Add Repair Policy cloud provider interface (#7345)
Browse files Browse the repository at this point in the history
  • Loading branch information
engedaam authored Nov 19, 2024
1 parent a4240fc commit 469ef7f
Show file tree
Hide file tree
Showing 10 changed files with 201 additions and 4 deletions.
1 change: 1 addition & 0 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ HELM_OPTS ?= --set serviceAccount.annotations.eks\\.amazonaws\\.com/role-arn=${K
--set controller.resources.limits.cpu=1 \
--set controller.resources.limits.memory=1Gi \
--set settings.featureGates.spotToSpotConsolidation=true \
--set settings.featureGates.nodeRepair=true \
--create-namespace

# CR for local builds of Karpenter
Expand Down
2 changes: 1 addition & 1 deletion charts/karpenter/templates/deployment.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ spec:
divisor: "0"
resource: limits.memory
- name: FEATURE_GATES
value: "SpotToSpotConsolidation={{ .Values.settings.featureGates.spotToSpotConsolidation }}"
value: "SpotToSpotConsolidation={{ .Values.settings.featureGates.spotToSpotConsolidation }},NodeRepair={{ .Values.settings.featureGates.nodeRepair }}"
{{- with .Values.settings.batchMaxDuration }}
- name: BATCH_MAX_DURATION
value: "{{ . }}"
Expand Down
3 changes: 3 additions & 0 deletions charts/karpenter/values.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -181,3 +181,6 @@ settings:
# -- spotToSpotConsolidation is ALPHA and is disabled by default.
# Setting this to true will enable spot replacement consolidation for both single and multi-node consolidation.
spotToSpotConsolidation: false
# -- nodeRepair is ALPHA and is disabled by default.
# Setting this to true will enable node repair.
nodeRepair: false
1 change: 1 addition & 0 deletions cmd/controller/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ func main() {

op.
WithControllers(ctx, corecontrollers.NewControllers(
ctx,
op.Manager,
op.Clock,
op.GetClient(),
Expand Down
2 changes: 1 addition & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ require (
k8s.io/klog/v2 v2.130.1
k8s.io/utils v0.0.0-20240711033017-18e509b52bc8
sigs.k8s.io/controller-runtime v0.19.1
sigs.k8s.io/karpenter v1.0.1-0.20241115180652-995040b20d0a
sigs.k8s.io/karpenter v1.0.1-0.20241119070053-a03600e47f44
sigs.k8s.io/yaml v1.4.0
)

Expand Down
4 changes: 2 additions & 2 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -323,8 +323,8 @@ sigs.k8s.io/controller-runtime v0.19.1 h1:Son+Q40+Be3QWb+niBXAg2vFiYWolDjjRfO8hn
sigs.k8s.io/controller-runtime v0.19.1/go.mod h1:iRmWllt8IlaLjvTTDLhRBXIEtkCK6hwVBJJsYS9Ajf4=
sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd h1:EDPBXCAspyGV4jQlpZSudPeMmr1bNJefnuqLsRAsHZo=
sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0=
sigs.k8s.io/karpenter v1.0.1-0.20241115180652-995040b20d0a h1:fbD7tYsZCIu3uzgJKO9/XWS6uYPCeWxQmwAvP2jtXRE=
sigs.k8s.io/karpenter v1.0.1-0.20241115180652-995040b20d0a/go.mod h1:zolnK/3MxqSPEhEan2VBbzuGdReJPFTbpYWGivwTgic=
sigs.k8s.io/karpenter v1.0.1-0.20241119070053-a03600e47f44 h1:s8UHRXbVIfVXoyYG/er0rQaMY75wZYWlgzI/Tlzp68U=
sigs.k8s.io/karpenter v1.0.1-0.20241119070053-a03600e47f44/go.mod h1:zolnK/3MxqSPEhEan2VBbzuGdReJPFTbpYWGivwTgic=
sigs.k8s.io/structured-merge-diff/v4 v4.4.1 h1:150L+0vs/8DA78h1u02ooW1/fFq/Lwr+sGiqlzvrtq4=
sigs.k8s.io/structured-merge-diff/v4 v4.4.1/go.mod h1:N8hJocpFajUSSeSJ9bOZ77VzejKZaXsTtZo4/u7Io08=
sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=
Expand Down
21 changes: 21 additions & 0 deletions pkg/cloudprovider/cloudprovider.go
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,27 @@ func getTags(ctx context.Context, nodeClass *v1.EC2NodeClass, nodeClaim *karpv1.
}), staticTags)
}

func (c *CloudProvider) RepairPolicies() []cloudprovider.RepairPolicy {
return []cloudprovider.RepairPolicy{
// Supported Kubelet fields
{
ConditionType: corev1.NodeReady,
ConditionStatus: corev1.ConditionFalse,
TolerationDuration: 30 * time.Minute,
},
{
ConditionType: corev1.NodeDiskPressure,
ConditionStatus: corev1.ConditionTrue,
TolerationDuration: 30 * time.Minute,
},
{
ConditionType: corev1.NodeMemoryPressure,
ConditionStatus: corev1.ConditionTrue,
TolerationDuration: 30 * time.Minute,
},
}
}

func (c *CloudProvider) resolveNodeClassFromNodeClaim(ctx context.Context, nodeClaim *karpv1.NodeClaim) (*v1.EC2NodeClass, error) {
nodeClass := &v1.EC2NodeClass{}
if err := c.kubeClient.Get(ctx, types.NamespacedName{Name: nodeClaim.Spec.NodeClassRef.Name}, nodeClass); err != nil {
Expand Down
4 changes: 4 additions & 0 deletions pkg/fake/cloudprovider.go
Original file line number Diff line number Diff line change
Expand Up @@ -89,3 +89,7 @@ func (c *CloudProvider) Name() string {
func (c *CloudProvider) GetSupportedNodeClasses() []status.Object {
return []status.Object{&v1.EC2NodeClass{}}
}

func (c *CloudProvider) RepairPolicies() []corecloudprovider.RepairPolicy {
return []corecloudprovider.RepairPolicy{}
}
30 changes: 30 additions & 0 deletions test/pkg/environment/common/expectations.go
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,36 @@ func (env *Environment) ExpectUpdated(objects ...client.Object) {
}
}

// ExpectStatusUpdated will update objects in the cluster to match the inputs.
// WARNING: This ignores the resource version check, which can result in
// overwriting changes made by other controllers in the cluster.
// This is useful in ensuring that we can clean up resources by patching
// out finalizers.
// Grab the object before making the updates to reduce the chance of this race.
func (env *Environment) ExpectStatusUpdated(objects ...client.Object) {
GinkgoHelper()
for _, o := range objects {
Eventually(func(g Gomega) {
current := o.DeepCopyObject().(client.Object)
g.Expect(env.Client.Get(env.Context, client.ObjectKeyFromObject(current), current)).To(Succeed())
if current.GetResourceVersion() != o.GetResourceVersion() {
log.FromContext(env).Info(fmt.Sprintf("detected an update to an object (%s) with an outdated resource version, did you get the latest version of the object before patching?", lo.Must(apiutil.GVKForObject(o, env.Client.Scheme()))))
}
o.SetResourceVersion(current.GetResourceVersion())
g.Expect(env.Client.Status().Update(env.Context, o)).To(Succeed())
}).WithTimeout(time.Second * 10).Should(Succeed())
}
}

func ReplaceNodeConditions(node *corev1.Node, conds ...corev1.NodeCondition) *corev1.Node {
keys := sets.New[string](lo.Map(conds, func(c corev1.NodeCondition, _ int) string { return string(c.Type) })...)
node.Status.Conditions = lo.Reject(node.Status.Conditions, func(c corev1.NodeCondition, _ int) bool {
return keys.Has(string(c.Type))
})
node.Status.Conditions = append(node.Status.Conditions, conds...)
return node
}

// ExpectCreatedOrUpdated can update objects in the cluster to match the inputs.
// WARNING: ExpectUpdated ignores the resource version check, which can result in
// overwriting changes made by other controllers in the cluster.
Expand Down
137 changes: 137 additions & 0 deletions test/suites/integration/repair_policy_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,137 @@
/*
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 integration_test

import (
"time"

appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
karpenterv1 "sigs.k8s.io/karpenter/pkg/apis/v1"
coretest "sigs.k8s.io/karpenter/pkg/test"

"github.com/aws/karpenter-provider-aws/test/pkg/environment/common"

. "github.com/onsi/ginkgo/v2"
"github.com/samber/lo"
)

var _ = Describe("Repair Policy", func() {
var selector labels.Selector
var dep *appsv1.Deployment
var numPods int
var unhealthyCondition corev1.NodeCondition

BeforeEach(func() {
unhealthyCondition = corev1.NodeCondition{
Type: corev1.NodeReady,
Status: corev1.ConditionFalse,
LastTransitionTime: metav1.Time{Time: time.Now().Add(-31 * time.Minute)},
}
numPods = 1
// Add pods with a do-not-disrupt annotation so that we can check node metadata before we disrupt
dep = coretest.Deployment(coretest.DeploymentOptions{
Replicas: int32(numPods),
PodOptions: coretest.PodOptions{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"app": "my-app",
},
Annotations: map[string]string{
karpenterv1.DoNotDisruptAnnotationKey: "true",
},
},
TerminationGracePeriodSeconds: lo.ToPtr[int64](0),
},
})
selector = labels.SelectorFromSet(dep.Spec.Selector.MatchLabels)
})

DescribeTable("Conditions", func(unhealthyCondition corev1.NodeCondition) {
env.ExpectCreated(nodeClass, nodePool, dep)
pod := env.EventuallyExpectHealthyPodCount(selector, numPods)[0]
node := env.ExpectCreatedNodeCount("==", 1)[0]
env.EventuallyExpectInitializedNodeCount("==", 1)

node = common.ReplaceNodeConditions(node, unhealthyCondition)
env.ExpectStatusUpdated(node)

env.EventuallyExpectNotFound(pod, node)
env.EventuallyExpectHealthyPodCount(selector, numPods)
},
Entry("Readiness", corev1.NodeCondition{
Type: corev1.NodeReady,
Status: corev1.ConditionFalse,
LastTransitionTime: metav1.Time{Time: time.Now().Add(-31 * time.Minute)},
}),
Entry("DiskPressure", corev1.NodeCondition{
Type: corev1.NodeDiskPressure,
Status: corev1.ConditionTrue,
LastTransitionTime: metav1.Time{Time: time.Now().Add(-31 * time.Minute)},
}),
Entry("MemoryPressure", corev1.NodeCondition{
Type: corev1.NodeMemoryPressure,
Status: corev1.ConditionTrue,
LastTransitionTime: metav1.Time{Time: time.Now().Add(-31 * time.Minute)},
}),
)
It("should ignore disruption budgets", func() {
nodePool.Spec.Disruption.Budgets = []karpenterv1.Budget{
{
Nodes: "0",
},
}
env.ExpectCreated(nodeClass, nodePool, dep)
pod := env.EventuallyExpectHealthyPodCount(selector, numPods)[0]
node := env.ExpectCreatedNodeCount("==", 1)[0]
env.EventuallyExpectInitializedNodeCount("==", 1)

node = common.ReplaceNodeConditions(node, unhealthyCondition)
env.ExpectStatusUpdated(node)

env.EventuallyExpectNotFound(pod, node)
env.EventuallyExpectHealthyPodCount(selector, numPods)
})
It("should ignore do-not-disrupt annotation on node", func() {
env.ExpectCreated(nodeClass, nodePool, dep)
pod := env.EventuallyExpectHealthyPodCount(selector, numPods)[0]
node := env.ExpectCreatedNodeCount("==", 1)[0]
env.EventuallyExpectInitializedNodeCount("==", 1)

node.Annotations[karpenterv1.DoNotDisruptAnnotationKey] = "true"
env.ExpectUpdated(node)

node = common.ReplaceNodeConditions(node, unhealthyCondition)
env.ExpectStatusUpdated(node)

env.EventuallyExpectNotFound(pod, node)
env.EventuallyExpectHealthyPodCount(selector, numPods)
})
It("should ignore terminationGracePeriod on the nodepool", func() {
nodePool.Spec.Template.Spec.TerminationGracePeriod = &metav1.Duration{Duration: time.Hour}
env.ExpectCreated(nodeClass, nodePool, dep)
pod := env.EventuallyExpectHealthyPodCount(selector, numPods)[0]
node := env.ExpectCreatedNodeCount("==", 1)[0]
env.EventuallyExpectInitializedNodeCount("==", 1)

node = common.ReplaceNodeConditions(node, unhealthyCondition)
env.ExpectStatusUpdated(node)

env.EventuallyExpectNotFound(pod, node)
env.EventuallyExpectHealthyPodCount(selector, numPods)
})
})

0 comments on commit 469ef7f

Please sign in to comment.