upstream/kubernetes
1
0
Fork 0
mirror of https://github.com/kubernetes/kubernetes.git synced 2026-05-28 04:04:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

controller/disruption: Improve goroutine mgmt

Browse source 
Make sure all threads are terminated when Run returns.
This commit is contained in:
Ondra Kupka 2025-10-27 12:10:07 +01:00
parent 6e0a4da2f6
commit d9ba92ba3b
1 changed files with 19 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
pkg/controller/disruption/disruption.go
View file

@ -19,6 +19,7 @@ package disruption
import (
"context"
"fmt"
"sync"
"time"
apps "k8s.io/api/apps/v1beta1"
@ -461,21 +462,30 @@ func (dc *DisruptionController) Run(ctx context.Context) {
}
defer dc.broadcaster.Shutdown()
defer dc.queue.ShutDown()
defer dc.recheckQueue.ShutDown()
defer dc.stalePodDisruptionQueue.ShutDown()
logger.Info("Starting disruption controller")
defer logger.Info("Shutting down disruption controller")
var wg sync.WaitGroup
defer func() {
logger.Info("Shutting down disruption controller")
dc.queue.ShutDown()
dc.recheckQueue.ShutDown()
dc.stalePodDisruptionQueue.ShutDown()
wg.Wait()
}()
if !cache.WaitForNamedCacheSyncWithContext(ctx, dc.podListerSynced, dc.pdbListerSynced, dc.rcListerSynced, dc.rsListerSynced, dc.dListerSynced, dc.ssListerSynced) {
return
}
go wait.UntilWithContext(ctx, dc.worker, time.Second)
go wait.Until(dc.recheckWorker, time.Second, ctx.Done())
go wait.UntilWithContext(ctx, dc.stalePodDisruptionWorker, time.Second)
wg.Go(func() {
wait.UntilWithContext(ctx, dc.worker, time.Second)
})
wg.Go(func() {
wait.Until(dc.recheckWorker, time.Second, ctx.Done())
})
wg.Go(func() {
wait.UntilWithContext(ctx, dc.stalePodDisruptionWorker, time.Second)
})
<-ctx.Done()
}