github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 02:06:23 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #132620 from tchap/kcm-clean-termination

Browse Source 
kube-controller-manager: Refactor leader election management to prepare for releasing lock on exit
This commit is contained in:
Kubernetes Prow Robot
2026-03-13 03:49:38 +05:30
committed by GitHub
parent 4730b179bd 8ddefcf4b3
commit d73c1818e9
1 changed files with 84 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
cmd/kube-controller-manager/app/controllermanager.go
View File

@@ -21,6 +21,7 @@ package app
import (
"context"
"errors"
"fmt"
"math/rand"
"net/http"
@@ -35,10 +36,12 @@ import (
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
utilerrors "k8s.io/apimachinery/pkg/util/errors"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/uuid"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apiserver/pkg/server"
"k8s.io/apiserver/pkg/server/flagz"
"k8s.io/apiserver/pkg/server/healthz"
"k8s.io/apiserver/pkg/server/mux"
@@ -137,6 +140,9 @@ controller, and serviceaccounts controller.`,
}
cliflag.PrintFlags(cmd.Flags())
// We use context.Background() here still because using server.SetupSignalContext() would cause
// components like the event broadcaster to terminate on signal immediately, which is not what we want.
// Termination for that case is being handled explicitly in Run() later on.
ctx := context.Background()
c, err := s.Config(ctx, KnownControllers(), ControllersDisabledByDefault(), ControllerAliases())
if err != nil {
@@ -148,7 +154,7 @@ controller, and serviceaccounts controller.`,
fg.(featuregate.MutableFeatureGate).AddMetrics()
// add component version metrics
s.ComponentGlobalsRegistry.AddMetrics()
return Run(ctx, c.Complete())
return Run(server.SetupSignalContext(), c.Complete())
},
Args: func(cmd *cobra.Command, args []string) error {
for _, arg := range args {
@@ -254,18 +260,18 @@ func Run(ctx context.Context, c *config.CompletedConfig) error {
saTokenControllerDescriptor := newServiceAccountTokenControllerDescriptor(rootClientBuilder)
run := func(ctx context.Context, controllerDescriptors map[string]*ControllerDescriptor) {
run := func(ctx context.Context, controllerDescriptors map[string]*ControllerDescriptor) error {
controllerContext, err := CreateControllerContext(ctx, c, rootClientBuilder, clientBuilder)
if err != nil {
logger.Error(err, "Error building controller context")
klog.FlushAndExit(klog.ExitFlushTimeout, 1)
return err
}
// Prepare all controllers in advance.
controllers, err := BuildControllers(ctx, controllerContext, controllerDescriptors, unsecuredMux, healthzHandler)
if err != nil {
logger.Error(err, "Error building controllers")
klog.FlushAndExit(klog.ExitFlushTimeout, 1)
return err
}
// Start the informers.
@@ -278,19 +284,19 @@ func Run(ctx context.Context, c *config.CompletedConfig) error {
// Actually start the controllers.
if len(controllers) > 0 {
if !RunControllers(ctx, controllerContext, controllers, ControllerStartJitter, c.ControllerShutdownTimeout) {
klog.FlushAndExit(klog.ExitFlushTimeout, 1)
return errors.New("controller shutdown timeout reached")
}
} else {
<-ctx.Done()
}
return nil
}
// No leader election, run directly
if !c.ComponentConfig.Generic.LeaderElection.LeaderElect {
controllerDescriptors := NewControllerDescriptors()
controllerDescriptors[names.ServiceAccountTokenController] = saTokenControllerDescriptor
run(ctx, controllerDescriptors)
return nil
return run(ctx, controllerDescriptors)
}
id, err := os.Hostname()
@@ -359,27 +365,50 @@ func Run(ctx context.Context, c *config.CompletedConfig) error {
go leaseCandidate.Run(ctx)
}
// Start the main lock
go leaderElectAndRun(ctx, c, id, electionChecker,
c.ComponentConfig.Generic.LeaderElection.ResourceLock,
c.ComponentConfig.Generic.LeaderElection.ResourceName,
leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
controllerDescriptors := NewControllerDescriptors()
if leaderMigrator != nil {
// If leader migration is enabled, we should start only non-migrated controllers
// for the main lock.
controllerDescriptors = filteredControllerDescriptors(controllerDescriptors, leaderMigrator.FilterFunc, leadermigration.ControllerNonMigrated)
logger.Info("leader migration: starting main controllers.")
}
controllerDescriptors[names.ServiceAccountTokenController] = saTokenControllerDescriptor
run(ctx, controllerDescriptors)
},
OnStoppedLeading: func() {
logger.Error(nil, "leaderelection lost")
klog.FlushAndExit(klog.ExitFlushTimeout, 1)
},
})
// Start the main lock.
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// startedLeading must be used to wrap any OnStartedLeading leader election callback.
var (
errs []error
errsLock sync.Mutex
)
startedLeading := func(next func(context.Context) error) func(context.Context) {
return func(ctx context.Context) {
// It's more efficient to cancel the context at the end of OnStartedLeading to signal termination,
// because OnStoppedLeading is only called once the LE lock is released.
defer cancel()
if err := next(ctx); err != nil && !errors.Is(err, context.Canceled) {
errsLock.Lock()
errs = append(errs, err)
errsLock.Unlock()
}
}
}
var wg sync.WaitGroup
wg.Go(func() {
leaderElectAndRun(ctx, c, id, electionChecker,
c.ComponentConfig.Generic.LeaderElection.ResourceLock,
c.ComponentConfig.Generic.LeaderElection.ResourceName,
leaderelection.LeaderCallbacks{
OnStartedLeading: startedLeading(func(ctx context.Context) error {
controllerDescriptors := NewControllerDescriptors()
if leaderMigrator != nil {
// If leader migration is enabled, we should start only non-migrated controllers
// for the main lock.
controllerDescriptors = filteredControllerDescriptors(controllerDescriptors, leaderMigrator.FilterFunc, leadermigration.ControllerNonMigrated)
logger.Info("leader migration: starting main controllers.")
}
controllerDescriptors[names.ServiceAccountTokenController] = saTokenControllerDescriptor
return run(ctx, controllerDescriptors)
}),
OnStoppedLeading: func() {
logger.Error(nil, "leaderelection lost/stopped")
},
})
})
// If Leader Migration is enabled, proceed to attempt the migration lock.
if leaderMigrator != nil {
@@ -387,30 +416,36 @@ func Run(ctx context.Context, c *config.CompletedConfig) error {
// At this point, the main lock must have already been acquired, or the KCM process already exited.
// We wait for the main lock before acquiring the migration lock to prevent the situation
// where KCM instance A holds the main lock while KCM instance B holds the migration lock.
<-leaderMigrator.MigrationReady
// Start the migration lock.
go leaderElectAndRun(ctx, c, id, electionChecker,
c.ComponentConfig.Generic.LeaderMigration.ResourceLock,
c.ComponentConfig.Generic.LeaderMigration.LeaderName,
leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
logger.Info("leader migration: starting migrated controllers.")
controllerDescriptors := NewControllerDescriptors()
controllerDescriptors = filteredControllerDescriptors(controllerDescriptors, leaderMigrator.FilterFunc, leadermigration.ControllerMigrated)
// DO NOT start saTokenController under migration lock
delete(controllerDescriptors, names.ServiceAccountTokenController)
run(ctx, controllerDescriptors)
},
OnStoppedLeading: func() {
logger.Error(nil, "migration leaderelection lost")
klog.FlushAndExit(klog.ExitFlushTimeout, 1)
},
select {
case <-leaderMigrator.MigrationReady:
// Start the migration lock.
wg.Go(func() {
leaderElectAndRun(ctx, c, id, electionChecker,
c.ComponentConfig.Generic.LeaderMigration.ResourceLock,
c.ComponentConfig.Generic.LeaderMigration.LeaderName,
leaderelection.LeaderCallbacks{
OnStartedLeading: startedLeading(func(ctx context.Context) error {
logger.Info("leader migration: starting migrated controllers.")
controllerDescriptors := NewControllerDescriptors()
controllerDescriptors = filteredControllerDescriptors(controllerDescriptors, leaderMigrator.FilterFunc, leadermigration.ControllerMigrated)
// DO NOT start saTokenController under migration lock
delete(controllerDescriptors, names.ServiceAccountTokenController)
return run(ctx, controllerDescriptors)
}),
OnStoppedLeading: func() {
logger.Error(nil, "migration leaderelection lost/stopped")
},
})
})
case <-ctx.Done():
}
}
<-stopCh
return nil
// Block until all leader elections are stopped.
wg.Wait()
// There is no need to hold errsLock since by this time all goroutines have terminated.
return utilerrors.NewAggregate(errs)
}
// ControllerContext defines the context object for controller
@@ -817,8 +852,6 @@ func leaderElectAndRun(ctx context.Context, c *config.CompletedConfig, lockIdent
Name: leaseName,
Coordinated: utilfeature.DefaultFeatureGate.Enabled(kubefeatures.CoordinatedLeaderElection),
})
panic("unreachable")
}
// filteredControllerDescriptors returns all controllerDescriptors after filtering through filterFunc.