Skip to content

perf: Refactor the eviction queue to be multithreaded #2252

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 14 commits into from
Jun 12, 2025
Merged

perf: Refactor the eviction queue to be multithreaded #2252

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
062e604
feat: Refactor the eviction queue to be multithreaded
DerekFrank May 23, 2025
b66a6c2
testing-changes
DerekFrank Jun 3, 2025
f456b13
Merge branch 'main' into threaded-eviction-queue3
DerekFrank Jun 3, 2025
76ccad5
blah
DerekFrank Jun 3, 2025
4873fcb
testing-changes
DerekFrank Jun 3, 2025
7ab99b9
feedback
DerekFrank Jun 3, 2025
afa8667
bigger buffered channel and no goroutine
DerekFrank Jun 3, 2025
68ff59f
passing test
DerekFrank Jun 4, 2025
51b3df7
minor feedback
DerekFrank Jun 5, 2025
3ac4431
Additional review feedback
DerekFrank Jun 10, 2025
a9e06fc
Additional review feedback
DerekFrank Jun 10, 2025
cb55f4a
Additional Review Feedback
DerekFrank Jun 11, 2025
f11cdc0
Removing the .Evict() function
DerekFrank Jun 12, 2025
718e558
minor feedback
DerekFrank Jun 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pkg/controllers/node/health/suite_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ var _ = BeforeSuite(func() {
cloudProvider = fake.NewCloudProvider()
cloudProvider = fake.NewCloudProvider()
recorder = test.NewEventRecorder()
queue = terminator.NewTestingQueue(env.Client, recorder)
queue = terminator.NewQueue(env.Client, recorder)
healthController = health.NewController(env.Client, cloudProvider, fakeClock, recorder)
})

Expand Down
101 changes: 43 additions & 58 deletions pkg/controllers/node/termination/suite_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ var _ = BeforeSuite(func() {

cloudProvider = fake.NewCloudProvider()
recorder = test.NewEventRecorder()
queue = terminator.NewTestingQueue(env.Client, recorder)
queue = terminator.NewQueue(env.Client, recorder)
terminationController = termination.NewController(fakeClock, env.Client, cloudProvider, terminator.NewTerminator(fakeClock, env.Client, queue, recorder), recorder)
})

Expand All @@ -86,7 +86,7 @@ var _ = Describe("Termination", func() {
BeforeEach(func() {
fakeClock.SetTime(time.Now())
cloudProvider.Reset()
*queue = lo.FromPtr(terminator.NewTestingQueue(env.Client, recorder))
*queue = lo.FromPtr(terminator.NewQueue(env.Client, recorder))

nodePool = test.NodePool()
nodeClaim, node = test.NodeClaimAndNode(v1.NodeClaim{ObjectMeta: metav1.ObjectMeta{Finalizers: []string{v1.TerminationFinalizer}}})
Expand Down Expand Up @@ -209,7 +209,8 @@ var _ = Describe("Termination", func() {
ExpectManualBinding(ctx, env.Client, pod, node)
Expect(env.Client.Delete(ctx, node)).To(Succeed())
// We only reconcile once since this label should be applied before draining the node
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node))
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation

node = ExpectNodeExists(ctx, env.Client, node.Name)
Expect(node.Labels[corev1.LabelNodeExcludeBalancers]).Should(Equal("karpenter"))
})
Expand All @@ -225,9 +226,10 @@ var _ = Describe("Termination", func() {
// Trigger Termination Controller
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)

ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
Expect(queue.Has(node, podSkip)).To(BeFalse())
ExpectSingletonReconciled(ctx, queue)
Expect(queue.Has(podSkip)).To(BeFalse())
ExpectObjectReconciled(ctx, env.Client, queue, podEvict)

// Expect node to exist and be draining
ExpectNodeWithNodeClaimDraining(env.Client, node.Name)
Expand Down Expand Up @@ -256,9 +258,10 @@ var _ = Describe("Termination", func() {
// Trigger Termination Controller
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)

ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
Expect(queue.Has(node, podSkip)).To(BeFalse())
ExpectSingletonReconciled(ctx, queue)
Expect(queue.Has(podSkip)).To(BeFalse())
ExpectObjectReconciled(ctx, env.Client, queue, podEvict)

// Expect node to exist and be draining
ExpectNodeWithNodeClaimDraining(env.Client, node.Name)
Expand All @@ -267,7 +270,7 @@ var _ = Describe("Termination", func() {
EventuallyExpectTerminating(ctx, env.Client, podEvict)
ExpectDeleted(ctx, env.Client, podEvict)

Expect(queue.Has(node, podSkip)).To(BeFalse())
Expect(queue.Has(podSkip)).To(BeFalse())

// Reconcile to delete node
node = ExpectNodeExists(ctx, env.Client, node.Name)
Expand All @@ -289,7 +292,7 @@ var _ = Describe("Termination", func() {
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, podEvict)

// Expect node to exist and be draining
ExpectNodeWithNodeClaimDraining(env.Client, node.Name)
Expand Down Expand Up @@ -317,8 +320,9 @@ var _ = Describe("Termination", func() {
ExpectApplied(ctx, env.Client, node, nodeClaim, pod)
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)

ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, pod)

// Expect pod with no owner ref to be enqueued for eviction
EventuallyExpectTerminating(ctx, env.Client, pod)
Expand Down Expand Up @@ -385,13 +389,13 @@ var _ = Describe("Termination", func() {
ExpectNodeWithNodeClaimDraining(env.Client, node.Name)

// Expect podNoEvict to be added to the queue
Expect(queue.Has(node, podNoEvict)).To(BeTrue())
Expect(queue.Has(podNoEvict)).To(BeTrue())

// Attempt to evict the pod, but fail to do so
ExpectSingletonReconciled(ctx, queue)
_ = ExpectObjectReconcileFailed(ctx, env.Client, queue, podNoEvict)

// Expect podNoEvict to fail eviction due to PDB, and be retried
Expect(queue.Has(node, podNoEvict)).To(BeTrue())
Expect(queue.Has(podNoEvict)).To(BeTrue())

// Delete pod to simulate successful eviction
ExpectDeleted(ctx, env.Client, podNoEvict)
Expand Down Expand Up @@ -452,18 +456,17 @@ var _ = Describe("Termination", func() {
}
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node))
ExpectNodeWithNodeClaimDraining(env.Client, node.Name)
for range podGroup {
ExpectSingletonReconciled(ctx, queue)
for _, pod := range podGroup {
ExpectObjectReconciled(ctx, env.Client, queue, pod)
}
// Start draining the pod group, but don't complete it yet
EventuallyExpectTerminating(ctx, env.Client, lo.Map(podGroup, func(p *corev1.Pod, _ int) client.Object { return p })...)

// Look at the next pod group and ensure that none of the pods have started terminating on it
if i != len(podGroups)-1 {
for range podGroups[i+1] {
ExpectSingletonReconciled(ctx, queue)
for _, pod := range podGroups[i+1] {
Expect(queue.Has(pod)).To(BeFalse())
}
ConsistentlyExpectNotTerminating(ctx, env.Client, lo.Map(podGroups[i+1], func(p *corev1.Pod, _ int) client.Object { return p })...)
}
// Expect that the pods are deleted -- which should unblock the next pod group
ExpectDeleted(ctx, env.Client, lo.Map(podGroup, func(p *corev1.Pod, _ int) client.Object { return p })...)
Expand All @@ -487,8 +490,9 @@ var _ = Describe("Termination", func() {
// Trigger Termination Controller
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)

ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation (non-critical)
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, podEvict)

// Expect node to exist and be draining
ExpectNodeWithNodeClaimDraining(env.Client, node.Name)
Expand All @@ -500,8 +504,8 @@ var _ = Describe("Termination", func() {
// Expect the critical pods to be evicted and deleted
node = ExpectNodeExists(ctx, env.Client, node.Name)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation (critical)
ExpectSingletonReconciled(ctx, queue)
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, podNodeCritical)
ExpectObjectReconciled(ctx, env.Client, queue, podClusterCritical)

EventuallyExpectTerminating(ctx, env.Client, podNodeCritical, podClusterCritical)
ExpectDeleted(ctx, env.Client, podNodeCritical)
Expand Down Expand Up @@ -535,10 +539,10 @@ var _ = Describe("Termination", func() {
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, podEvict)

// Expect mirror pod to not be queued for eviction
Expect(queue.Has(node, podNoEvict)).To(BeFalse())
Expect(queue.Has(podNoEvict)).To(BeFalse())

// Expect podEvict to be enqueued for eviction then be successful
EventuallyExpectTerminating(ctx, env.Client, podEvict)
Expand Down Expand Up @@ -566,8 +570,8 @@ var _ = Describe("Termination", func() {
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectSingletonReconciled(ctx, queue)
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, pods[0])
ExpectObjectReconciled(ctx, env.Client, queue, pods[1])

nodeClaim = ExpectExists(ctx, env.Client, nodeClaim)
Expect(nodeClaim.StatusConditions().Get(v1.ConditionTypeDrained).IsUnknown()).To(BeTrue())
Expand Down Expand Up @@ -614,8 +618,8 @@ var _ = Describe("Termination", func() {
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectSingletonReconciled(ctx, queue)
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, pods[0])
ExpectObjectReconciled(ctx, env.Client, queue, pods[1])

// Expect the pods to be evicted
EventuallyExpectTerminating(ctx, env.Client, pods[0], pods[1])
Expand Down Expand Up @@ -644,8 +648,8 @@ var _ = Describe("Termination", func() {
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectSingletonReconciled(ctx, queue)
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, pods[0])
ExpectObjectReconciled(ctx, env.Client, queue, pods[1])

// Expect the pods to be evicted
EventuallyExpectTerminating(ctx, env.Client, pods[0], pods[1])
Expand Down Expand Up @@ -675,7 +679,8 @@ var _ = Describe("Termination", func() {
// Before grace period, node should not delete
Expect(env.Client.Delete(ctx, node)).To(Succeed())
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectSingletonReconciled(ctx, queue)
ExpectObjectReconciled(ctx, env.Client, queue, pod)

ExpectNodeExists(ctx, env.Client, node.Name)
EventuallyExpectTerminating(ctx, env.Client, pod)

Expand Down Expand Up @@ -720,7 +725,7 @@ var _ = Describe("Termination", func() {
ExpectNotFound(ctx, env.Client, node)

// Expect that the old pod's key still exists in the queue
Expect(queue.Has(node, pod)).To(BeTrue())
Expect(queue.Has(pod)).To(BeTrue())

// Re-create the pod and node, it should now have the same name, but a different UUID
node = test.Node(test.NodeOptions{
Expand All @@ -737,8 +742,8 @@ var _ = Describe("Termination", func() {
})
ExpectApplied(ctx, env.Client, node, pod)

// Trigger eviction queue with the pod key still in it
ExpectSingletonReconciled(ctx, queue)
// Check if the queue has seen the pod (it shouldn't because its got a different UUID)
Expect(queue.Has(pod)).To(BeFalse())

Consistently(func(g Gomega) {
g.Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(pod), pod)).To(Succeed())
Expand All @@ -764,6 +769,8 @@ var _ = Describe("Termination", func() {

Expect(env.Client.Delete(ctx, node)).To(Succeed())
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // DrainInitiation
ExpectObjectReconciled(ctx, env.Client, queue, pod)

ExpectNodeExists(ctx, env.Client, node.Name)
pod = ExpectExists(ctx, env.Client, pod)
Expect(pod.DeletionTimestamp.IsZero()).To(BeFalse())
Expand Down Expand Up @@ -798,8 +805,9 @@ var _ = Describe("Termination", func() {
// The pod should be deleted 60 seconds before the node's TGP expires
fakeClock.Step(175 * time.Second)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node))
pod = ExpectExists(ctx, env.Client, pod)
Expect(pod.DeletionTimestamp.IsZero()).To(BeFalse())
ExpectObjectReconciled(ctx, env.Client, queue, pod)
ExpectDeleted(ctx, env.Client, pod)
Copy link
Member

@jonathan-innis jonathan-innis Jun 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Like I mentioned below -- this actually does a deletion, which I don't think is what we were doing before -- is this what we want? Since we aren't really checking anything here after reconciling


})
Context("VolumeAttachments", func() {
It("should wait for volume attachments", func() {
Expand Down Expand Up @@ -944,29 +952,6 @@ var _ = Describe("Termination", func() {
Expect(ok).To(BeTrue())
Expect(lo.FromPtr(m.GetHistogram().SampleCount)).To(BeNumerically("==", 1))
})
It("should update the eviction queueDepth metric when reconciling pods", func() {
minAvailable := intstr.FromInt32(0)
labelSelector := map[string]string{test.RandomName(): test.RandomName()}
pdb := test.PodDisruptionBudget(test.PDBOptions{
Labels: labelSelector,
// Don't let any pod evict
MinAvailable: &minAvailable,
})
ExpectApplied(ctx, env.Client, pdb, node, nodeClaim)
pods := test.Pods(5, test.PodOptions{NodeName: node.Name, ObjectMeta: metav1.ObjectMeta{
OwnerReferences: defaultOwnerRefs,
Labels: labelSelector,
}})
ExpectApplied(ctx, env.Client, lo.Map(pods, func(p *corev1.Pod, _ int) client.Object { return p })...)

wqDepthBefore, _ := FindMetricWithLabelValues("workqueue_adds_total", map[string]string{"name": "eviction.workqueue"})
Expect(env.Client.Delete(ctx, node)).To(Succeed())
node = ExpectNodeExists(ctx, env.Client, node.Name)
ExpectRequeued(ExpectObjectReconciled(ctx, env.Client, terminationController, node)) // Drain
wqDepthAfter, ok := FindMetricWithLabelValues("workqueue_adds_total", map[string]string{"name": "eviction.workqueue"})
Expect(ok).To(BeTrue())
Expect(lo.FromPtr(wqDepthAfter.GetCounter().Value) - lo.FromPtr(wqDepthBefore.GetCounter().Value)).To(BeNumerically("==", 5))
})
})
})

Expand Down
Loading
Loading