mirror of
https://github.com/k3s-io/k3s.git
synced 2024-06-07 19:41:36 +00:00
e8381db778
* Update Kubernetes to v1.21.0 * Update to golang v1.16.2 * Update dependent modules to track with upstream * Switch to upstream flannel * Track changes to upstream cloud-controller-manager and FeatureGates Signed-off-by: Brad Davidson <brad.davidson@rancher.com>
711 lines
29 KiB
Go
711 lines
29 KiB
Go
/*
|
|
Copyright 2014 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 app implements a server that runs a set of active
|
|
// components. This includes replication controllers, service endpoints and
|
|
// nodes.
|
|
//
|
|
package app
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io/ioutil"
|
|
"math/rand"
|
|
"net/http"
|
|
"os"
|
|
"time"
|
|
|
|
"github.com/spf13/cobra"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime/schema"
|
|
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"
|
|
genericfeatures "k8s.io/apiserver/pkg/features"
|
|
"k8s.io/apiserver/pkg/server"
|
|
"k8s.io/apiserver/pkg/server/healthz"
|
|
"k8s.io/apiserver/pkg/server/mux"
|
|
utilfeature "k8s.io/apiserver/pkg/util/feature"
|
|
cacheddiscovery "k8s.io/client-go/discovery/cached"
|
|
"k8s.io/client-go/informers"
|
|
"k8s.io/client-go/metadata"
|
|
"k8s.io/client-go/metadata/metadatainformer"
|
|
restclient "k8s.io/client-go/rest"
|
|
"k8s.io/client-go/restmapper"
|
|
"k8s.io/client-go/tools/leaderelection"
|
|
"k8s.io/client-go/tools/leaderelection/resourcelock"
|
|
certutil "k8s.io/client-go/util/cert"
|
|
"k8s.io/client-go/util/keyutil"
|
|
cloudprovider "k8s.io/cloud-provider"
|
|
cliflag "k8s.io/component-base/cli/flag"
|
|
"k8s.io/component-base/cli/globalflag"
|
|
"k8s.io/component-base/configz"
|
|
"k8s.io/component-base/term"
|
|
"k8s.io/component-base/version"
|
|
"k8s.io/component-base/version/verflag"
|
|
genericcontrollermanager "k8s.io/controller-manager/app"
|
|
"k8s.io/controller-manager/pkg/clientbuilder"
|
|
"k8s.io/controller-manager/pkg/informerfactory"
|
|
"k8s.io/controller-manager/pkg/leadermigration"
|
|
"k8s.io/klog/v2"
|
|
"k8s.io/kubernetes/cmd/kube-controller-manager/app/config"
|
|
"k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
|
|
kubectrlmgrconfig "k8s.io/kubernetes/pkg/controller/apis/config"
|
|
serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
|
|
"k8s.io/kubernetes/pkg/serviceaccount"
|
|
)
|
|
|
|
const (
|
|
// ControllerStartJitter is the Jitter used when starting controller managers
|
|
ControllerStartJitter = 1.0
|
|
// ConfigzName is the name used for register kube-controller manager /configz, same with GroupName.
|
|
ConfigzName = "kubecontrollermanager.config.k8s.io"
|
|
)
|
|
|
|
// ControllerLoopMode is the kube-controller-manager's mode of running controller loops that are cloud provider dependent
|
|
type ControllerLoopMode int
|
|
|
|
const (
|
|
// IncludeCloudLoops means the kube-controller-manager include the controller loops that are cloud provider dependent
|
|
IncludeCloudLoops ControllerLoopMode = iota
|
|
// ExternalLoops means the kube-controller-manager exclude the controller loops that are cloud provider dependent
|
|
ExternalLoops
|
|
)
|
|
|
|
// NewControllerManagerCommand creates a *cobra.Command object with default parameters
|
|
func NewControllerManagerCommand() *cobra.Command {
|
|
s, err := options.NewKubeControllerManagerOptions()
|
|
if err != nil {
|
|
klog.Fatalf("unable to initialize command options: %v", err)
|
|
}
|
|
|
|
cmd := &cobra.Command{
|
|
Use: "kube-controller-manager",
|
|
Long: `The Kubernetes controller manager is a daemon that embeds
|
|
the core control loops shipped with Kubernetes. In applications of robotics and
|
|
automation, a control loop is a non-terminating loop that regulates the state of
|
|
the system. In Kubernetes, a controller is a control loop that watches the shared
|
|
state of the cluster through the apiserver and makes changes attempting to move the
|
|
current state towards the desired state. Examples of controllers that ship with
|
|
Kubernetes today are the replication controller, endpoints controller, namespace
|
|
controller, and serviceaccounts controller.`,
|
|
PersistentPreRunE: func(*cobra.Command, []string) error {
|
|
// silence client-go warnings.
|
|
// kube-controller-manager generically watches APIs (including deprecated ones),
|
|
// and CI ensures it works properly against matching kube-apiserver versions.
|
|
restclient.SetDefaultWarningHandler(restclient.NoWarnings{})
|
|
return nil
|
|
},
|
|
Run: func(cmd *cobra.Command, args []string) {
|
|
verflag.PrintAndExitIfRequested()
|
|
cliflag.PrintFlags(cmd.Flags())
|
|
|
|
c, err := s.Config(KnownControllers(), ControllersDisabledByDefault.List())
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "%v\n", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
if err := Run(c.Complete(), wait.NeverStop); err != nil {
|
|
fmt.Fprintf(os.Stderr, "%v\n", err)
|
|
os.Exit(1)
|
|
}
|
|
},
|
|
Args: func(cmd *cobra.Command, args []string) error {
|
|
for _, arg := range args {
|
|
if len(arg) > 0 {
|
|
return fmt.Errorf("%q does not take any arguments, got %q", cmd.CommandPath(), args)
|
|
}
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
|
|
fs := cmd.Flags()
|
|
namedFlagSets := s.Flags(KnownControllers(), ControllersDisabledByDefault.List())
|
|
verflag.AddFlags(namedFlagSets.FlagSet("global"))
|
|
globalflag.AddGlobalFlags(namedFlagSets.FlagSet("global"), cmd.Name())
|
|
registerLegacyGlobalFlags(namedFlagSets)
|
|
for _, f := range namedFlagSets.FlagSets {
|
|
fs.AddFlagSet(f)
|
|
}
|
|
usageFmt := "Usage:\n %s\n"
|
|
cols, _, _ := term.TerminalSize(cmd.OutOrStdout())
|
|
cmd.SetUsageFunc(func(cmd *cobra.Command) error {
|
|
fmt.Fprintf(cmd.OutOrStderr(), usageFmt, cmd.UseLine())
|
|
cliflag.PrintSections(cmd.OutOrStderr(), namedFlagSets, cols)
|
|
return nil
|
|
})
|
|
cmd.SetHelpFunc(func(cmd *cobra.Command, args []string) {
|
|
fmt.Fprintf(cmd.OutOrStdout(), "%s\n\n"+usageFmt, cmd.Long, cmd.UseLine())
|
|
cliflag.PrintSections(cmd.OutOrStdout(), namedFlagSets, cols)
|
|
})
|
|
|
|
return cmd
|
|
}
|
|
|
|
// ResyncPeriod returns a function which generates a duration each time it is
|
|
// invoked; this is so that multiple controllers don't get into lock-step and all
|
|
// hammer the apiserver with list requests simultaneously.
|
|
func ResyncPeriod(c *config.CompletedConfig) func() time.Duration {
|
|
return func() time.Duration {
|
|
factor := rand.Float64() + 1
|
|
return time.Duration(float64(c.ComponentConfig.Generic.MinResyncPeriod.Nanoseconds()) * factor)
|
|
}
|
|
}
|
|
|
|
// Run runs the KubeControllerManagerOptions. This should never exit.
|
|
func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
|
|
// To help debugging, immediately log version
|
|
klog.Infof("Version: %+v", version.Get())
|
|
|
|
if cfgz, err := configz.New(ConfigzName); err == nil {
|
|
cfgz.Set(c.ComponentConfig)
|
|
} else {
|
|
klog.Errorf("unable to register configz: %v", err)
|
|
}
|
|
|
|
// Setup any healthz checks we will want to use.
|
|
var checks []healthz.HealthChecker
|
|
var electionChecker *leaderelection.HealthzAdaptor
|
|
if c.ComponentConfig.Generic.LeaderElection.LeaderElect {
|
|
electionChecker = leaderelection.NewLeaderHealthzAdaptor(time.Second * 20)
|
|
checks = append(checks, electionChecker)
|
|
}
|
|
|
|
// Start the controller manager HTTP server
|
|
// unsecuredMux is the handler for these controller *after* authn/authz filters have been applied
|
|
var unsecuredMux *mux.PathRecorderMux
|
|
if c.SecureServing != nil {
|
|
unsecuredMux = genericcontrollermanager.NewBaseHandler(&c.ComponentConfig.Generic.Debugging, checks...)
|
|
handler := genericcontrollermanager.BuildHandlerChain(unsecuredMux, &c.Authorization, &c.Authentication)
|
|
// TODO: handle stoppedCh returned by c.SecureServing.Serve
|
|
if _, err := c.SecureServing.Serve(handler, 0, stopCh); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if c.InsecureServing != nil {
|
|
unsecuredMux = genericcontrollermanager.NewBaseHandler(&c.ComponentConfig.Generic.Debugging, checks...)
|
|
insecureSuperuserAuthn := server.AuthenticationInfo{Authenticator: &server.InsecureSuperuser{}}
|
|
handler := genericcontrollermanager.BuildHandlerChain(unsecuredMux, nil, &insecureSuperuserAuthn)
|
|
if err := c.InsecureServing.Serve(handler, 0, stopCh); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
clientBuilder, rootClientBuilder := createClientBuilders(c)
|
|
|
|
saTokenControllerInitFunc := serviceAccountTokenControllerStarter{rootClientBuilder: rootClientBuilder}.startServiceAccountTokenController
|
|
|
|
run := func(ctx context.Context, startSATokenController InitFunc, initializersFunc ControllerInitializersFunc) {
|
|
|
|
controllerContext, err := CreateControllerContext(c, rootClientBuilder, clientBuilder, ctx.Done())
|
|
if err != nil {
|
|
klog.Fatalf("error building controller context: %v", err)
|
|
}
|
|
controllerInitializers := initializersFunc(controllerContext.LoopMode)
|
|
if err := StartControllers(controllerContext, startSATokenController, controllerInitializers, unsecuredMux); err != nil {
|
|
klog.Fatalf("error starting controllers: %v", err)
|
|
}
|
|
|
|
controllerContext.InformerFactory.Start(controllerContext.Stop)
|
|
controllerContext.ObjectOrMetadataInformerFactory.Start(controllerContext.Stop)
|
|
close(controllerContext.InformersStarted)
|
|
|
|
select {}
|
|
}
|
|
|
|
// No leader election, run directly
|
|
if !c.ComponentConfig.Generic.LeaderElection.LeaderElect {
|
|
run(context.TODO(), saTokenControllerInitFunc, NewControllerInitializers)
|
|
panic("unreachable")
|
|
}
|
|
|
|
id, err := os.Hostname()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// add a uniquifier so that two processes on the same host don't accidentally both become active
|
|
id = id + "_" + string(uuid.NewUUID())
|
|
|
|
// leaderMigrator will be non-nil if and only if Leader Migration is enabled.
|
|
var leaderMigrator *leadermigration.LeaderMigrator = nil
|
|
|
|
// startSATokenController will be original saTokenControllerInitFunc if leader migration is not enabled.
|
|
startSATokenController := saTokenControllerInitFunc
|
|
|
|
// If leader migration is enabled, create the LeaderMigrator and prepare for migration
|
|
if leadermigration.Enabled(&c.ComponentConfig.Generic) {
|
|
klog.Infof("starting leader migration")
|
|
|
|
leaderMigrator = leadermigration.NewLeaderMigrator(&c.ComponentConfig.Generic.LeaderMigration,
|
|
"kube-controller-manager")
|
|
|
|
// Wrap saTokenControllerInitFunc to signal readiness for migration after starting
|
|
// the controller.
|
|
startSATokenController = func(ctx ControllerContext) (http.Handler, bool, error) {
|
|
defer close(leaderMigrator.MigrationReady)
|
|
return saTokenControllerInitFunc(ctx)
|
|
}
|
|
}
|
|
|
|
// Start the main lock
|
|
go leaderElectAndRun(c, id, electionChecker,
|
|
c.ComponentConfig.Generic.LeaderElection.ResourceLock,
|
|
c.ComponentConfig.Generic.LeaderElection.ResourceName,
|
|
leaderelection.LeaderCallbacks{
|
|
OnStartedLeading: func(ctx context.Context) {
|
|
initializersFunc := NewControllerInitializers
|
|
if leaderMigrator != nil {
|
|
// If leader migration is enabled, we should start only non-migrated controllers
|
|
// for the main lock.
|
|
initializersFunc = createInitializersFunc(leaderMigrator.FilterFunc, leadermigration.ControllerNonMigrated)
|
|
klog.Info("leader migration: starting main controllers.")
|
|
}
|
|
run(ctx, startSATokenController, initializersFunc)
|
|
},
|
|
OnStoppedLeading: func() {
|
|
klog.Fatalf("leaderelection lost")
|
|
},
|
|
})
|
|
|
|
// If Leader Migration is enabled, proceed to attempt the migration lock.
|
|
if leaderMigrator != nil {
|
|
// Wait for Service Account Token Controller to start before acquiring the migration lock.
|
|
// 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(c, id, electionChecker,
|
|
c.ComponentConfig.Generic.LeaderMigration.ResourceLock,
|
|
c.ComponentConfig.Generic.LeaderMigration.LeaderName,
|
|
leaderelection.LeaderCallbacks{
|
|
OnStartedLeading: func(ctx context.Context) {
|
|
klog.Info("leader migration: starting migrated controllers.")
|
|
// DO NOT start saTokenController under migration lock
|
|
run(ctx, nil, createInitializersFunc(leaderMigrator.FilterFunc, leadermigration.ControllerMigrated))
|
|
},
|
|
OnStoppedLeading: func() {
|
|
klog.Fatalf("migration leaderelection lost")
|
|
},
|
|
})
|
|
}
|
|
|
|
select {}
|
|
}
|
|
|
|
// ControllerContext defines the context object for controller
|
|
type ControllerContext struct {
|
|
// ClientBuilder will provide a client for this controller to use
|
|
ClientBuilder clientbuilder.ControllerClientBuilder
|
|
|
|
// InformerFactory gives access to informers for the controller.
|
|
InformerFactory informers.SharedInformerFactory
|
|
|
|
// ObjectOrMetadataInformerFactory gives access to informers for typed resources
|
|
// and dynamic resources by their metadata. All generic controllers currently use
|
|
// object metadata - if a future controller needs access to the full object this
|
|
// would become GenericInformerFactory and take a dynamic client.
|
|
ObjectOrMetadataInformerFactory informerfactory.InformerFactory
|
|
|
|
// ComponentConfig provides access to init options for a given controller
|
|
ComponentConfig kubectrlmgrconfig.KubeControllerManagerConfiguration
|
|
|
|
// DeferredDiscoveryRESTMapper is a RESTMapper that will defer
|
|
// initialization of the RESTMapper until the first mapping is
|
|
// requested.
|
|
RESTMapper *restmapper.DeferredDiscoveryRESTMapper
|
|
|
|
// AvailableResources is a map listing currently available resources
|
|
AvailableResources map[schema.GroupVersionResource]bool
|
|
|
|
// Cloud is the cloud provider interface for the controllers to use.
|
|
// It must be initialized and ready to use.
|
|
Cloud cloudprovider.Interface
|
|
|
|
// Control for which control loops to be run
|
|
// IncludeCloudLoops is for a kube-controller-manager running all loops
|
|
// ExternalLoops is for a kube-controller-manager running with a cloud-controller-manager
|
|
LoopMode ControllerLoopMode
|
|
|
|
// Stop is the stop channel
|
|
Stop <-chan struct{}
|
|
|
|
// InformersStarted is closed after all of the controllers have been initialized and are running. After this point it is safe,
|
|
// for an individual controller to start the shared informers. Before it is closed, they should not.
|
|
InformersStarted chan struct{}
|
|
|
|
// ResyncPeriod generates a duration each time it is invoked; this is so that
|
|
// multiple controllers don't get into lock-step and all hammer the apiserver
|
|
// with list requests simultaneously.
|
|
ResyncPeriod func() time.Duration
|
|
}
|
|
|
|
// IsControllerEnabled checks if the context's controllers enabled or not
|
|
func (c ControllerContext) IsControllerEnabled(name string) bool {
|
|
return genericcontrollermanager.IsControllerEnabled(name, ControllersDisabledByDefault, c.ComponentConfig.Generic.Controllers)
|
|
}
|
|
|
|
// InitFunc is used to launch a particular controller. It may run additional "should I activate checks".
|
|
// Any error returned will cause the controller process to `Fatal`
|
|
// The bool indicates whether the controller was enabled.
|
|
type InitFunc func(ctx ControllerContext) (debuggingHandler http.Handler, enabled bool, err error)
|
|
|
|
// ControllerInitializersFunc is used to create a collection of initializers
|
|
// given the loopMode.
|
|
type ControllerInitializersFunc func(loopMode ControllerLoopMode) (initializers map[string]InitFunc)
|
|
|
|
var _ ControllerInitializersFunc = NewControllerInitializers
|
|
|
|
// KnownControllers returns all known controllers's name
|
|
func KnownControllers() []string {
|
|
ret := sets.StringKeySet(NewControllerInitializers(IncludeCloudLoops))
|
|
|
|
// add "special" controllers that aren't initialized normally. These controllers cannot be initialized
|
|
// using a normal function. The only known special case is the SA token controller which *must* be started
|
|
// first to ensure that the SA tokens for future controllers will exist. Think very carefully before adding
|
|
// to this list.
|
|
ret.Insert(
|
|
saTokenControllerName,
|
|
)
|
|
|
|
return ret.List()
|
|
}
|
|
|
|
// ControllersDisabledByDefault is the set of controllers which is disabled by default
|
|
var ControllersDisabledByDefault = sets.NewString(
|
|
"bootstrapsigner",
|
|
"tokencleaner",
|
|
)
|
|
|
|
const (
|
|
saTokenControllerName = "serviceaccount-token"
|
|
)
|
|
|
|
// NewControllerInitializers is a public map of named controller groups (you can start more than one in an init func)
|
|
// paired to their InitFunc. This allows for structured downstream composition and subdivision.
|
|
func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
|
|
controllers := map[string]InitFunc{}
|
|
controllers["endpoint"] = startEndpointController
|
|
controllers["endpointslice"] = startEndpointSliceController
|
|
controllers["endpointslicemirroring"] = startEndpointSliceMirroringController
|
|
controllers["replicationcontroller"] = startReplicationController
|
|
controllers["podgc"] = startPodGCController
|
|
controllers["resourcequota"] = startResourceQuotaController
|
|
controllers["namespace"] = startNamespaceController
|
|
controllers["serviceaccount"] = startServiceAccountController
|
|
controllers["garbagecollector"] = startGarbageCollectorController
|
|
controllers["daemonset"] = startDaemonSetController
|
|
controllers["job"] = startJobController
|
|
controllers["deployment"] = startDeploymentController
|
|
controllers["replicaset"] = startReplicaSetController
|
|
controllers["horizontalpodautoscaling"] = startHPAController
|
|
controllers["disruption"] = startDisruptionController
|
|
controllers["statefulset"] = startStatefulSetController
|
|
controllers["cronjob"] = startCronJobController
|
|
controllers["csrsigning"] = startCSRSigningController
|
|
controllers["csrapproving"] = startCSRApprovingController
|
|
controllers["csrcleaner"] = startCSRCleanerController
|
|
controllers["ttl"] = startTTLController
|
|
controllers["bootstrapsigner"] = startBootstrapSignerController
|
|
controllers["tokencleaner"] = startTokenCleanerController
|
|
controllers["nodeipam"] = startNodeIpamController
|
|
controllers["nodelifecycle"] = startNodeLifecycleController
|
|
if loopMode == IncludeCloudLoops {
|
|
controllers["service"] = startServiceController
|
|
controllers["route"] = startRouteController
|
|
controllers["cloud-node-lifecycle"] = startCloudNodeLifecycleController
|
|
// TODO: volume controller into the IncludeCloudLoops only set.
|
|
}
|
|
controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
|
|
controllers["attachdetach"] = startAttachDetachController
|
|
controllers["persistentvolume-expander"] = startVolumeExpandController
|
|
controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
|
|
controllers["pvc-protection"] = startPVCProtectionController
|
|
controllers["pv-protection"] = startPVProtectionController
|
|
controllers["ttl-after-finished"] = startTTLAfterFinishedController
|
|
controllers["root-ca-cert-publisher"] = startRootCACertPublisher
|
|
controllers["ephemeral-volume"] = startEphemeralVolumeController
|
|
if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerIdentity) &&
|
|
utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StorageVersionAPI) {
|
|
controllers["storage-version-gc"] = startStorageVersionGCController
|
|
}
|
|
|
|
return controllers
|
|
}
|
|
|
|
// GetAvailableResources gets the map which contains all available resources of the apiserver
|
|
// TODO: In general, any controller checking this needs to be dynamic so
|
|
// users don't have to restart their controller manager if they change the apiserver.
|
|
// Until we get there, the structure here needs to be exposed for the construction of a proper ControllerContext.
|
|
func GetAvailableResources(clientBuilder clientbuilder.ControllerClientBuilder) (map[schema.GroupVersionResource]bool, error) {
|
|
client := clientBuilder.ClientOrDie("controller-discovery")
|
|
discoveryClient := client.Discovery()
|
|
_, resourceMap, err := discoveryClient.ServerGroupsAndResources()
|
|
if err != nil {
|
|
utilruntime.HandleError(fmt.Errorf("unable to get all supported resources from server: %v", err))
|
|
}
|
|
if len(resourceMap) == 0 {
|
|
return nil, fmt.Errorf("unable to get any supported resources from server")
|
|
}
|
|
|
|
allResources := map[schema.GroupVersionResource]bool{}
|
|
for _, apiResourceList := range resourceMap {
|
|
version, err := schema.ParseGroupVersion(apiResourceList.GroupVersion)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, apiResource := range apiResourceList.APIResources {
|
|
allResources[version.WithResource(apiResource.Name)] = true
|
|
}
|
|
}
|
|
|
|
return allResources, nil
|
|
}
|
|
|
|
// CreateControllerContext creates a context struct containing references to resources needed by the
|
|
// controllers such as the cloud provider and clientBuilder. rootClientBuilder is only used for
|
|
// the shared-informers client and token controller.
|
|
func CreateControllerContext(s *config.CompletedConfig, rootClientBuilder, clientBuilder clientbuilder.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {
|
|
versionedClient := rootClientBuilder.ClientOrDie("shared-informers")
|
|
sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())
|
|
|
|
metadataClient := metadata.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("metadata-informers"))
|
|
metadataInformers := metadatainformer.NewSharedInformerFactory(metadataClient, ResyncPeriod(s)())
|
|
|
|
// If apiserver is not running we should wait for some time and fail only then. This is particularly
|
|
// important when we start apiserver and controller manager at the same time.
|
|
if err := genericcontrollermanager.WaitForAPIServer(versionedClient, 120*time.Second); err != nil {
|
|
return ControllerContext{}, fmt.Errorf("failed to wait for apiserver being healthy: %v", err)
|
|
}
|
|
|
|
// Use a discovery client capable of being refreshed.
|
|
discoveryClient := rootClientBuilder.DiscoveryClientOrDie("controller-discovery")
|
|
cachedClient := cacheddiscovery.NewMemCacheClient(discoveryClient)
|
|
restMapper := restmapper.NewDeferredDiscoveryRESTMapper(cachedClient)
|
|
go wait.Until(func() {
|
|
restMapper.Reset()
|
|
}, 30*time.Second, stop)
|
|
|
|
availableResources, err := GetAvailableResources(rootClientBuilder)
|
|
if err != nil {
|
|
return ControllerContext{}, err
|
|
}
|
|
|
|
cloud, loopMode, err := createCloudProvider(s.ComponentConfig.KubeCloudShared.CloudProvider.Name, s.ComponentConfig.KubeCloudShared.ExternalCloudVolumePlugin,
|
|
s.ComponentConfig.KubeCloudShared.CloudProvider.CloudConfigFile, s.ComponentConfig.KubeCloudShared.AllowUntaggedCloud, sharedInformers)
|
|
if err != nil {
|
|
return ControllerContext{}, err
|
|
}
|
|
|
|
ctx := ControllerContext{
|
|
ClientBuilder: clientBuilder,
|
|
InformerFactory: sharedInformers,
|
|
ObjectOrMetadataInformerFactory: informerfactory.NewInformerFactory(sharedInformers, metadataInformers),
|
|
ComponentConfig: s.ComponentConfig,
|
|
RESTMapper: restMapper,
|
|
AvailableResources: availableResources,
|
|
Cloud: cloud,
|
|
LoopMode: loopMode,
|
|
Stop: stop,
|
|
InformersStarted: make(chan struct{}),
|
|
ResyncPeriod: ResyncPeriod(s),
|
|
}
|
|
return ctx, nil
|
|
}
|
|
|
|
// StartControllers starts a set of controllers with a specified ControllerContext
|
|
func StartControllers(ctx ControllerContext, startSATokenController InitFunc, controllers map[string]InitFunc, unsecuredMux *mux.PathRecorderMux) error {
|
|
// Always start the SA token controller first using a full-power client, since it needs to mint tokens for the rest
|
|
// If this fails, just return here and fail since other controllers won't be able to get credentials.
|
|
if startSATokenController != nil {
|
|
if _, _, err := startSATokenController(ctx); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Initialize the cloud provider with a reference to the clientBuilder only after token controller
|
|
// has started in case the cloud provider uses the client builder.
|
|
if ctx.Cloud != nil {
|
|
ctx.Cloud.Initialize(ctx.ClientBuilder, ctx.Stop)
|
|
}
|
|
|
|
for controllerName, initFn := range controllers {
|
|
if !ctx.IsControllerEnabled(controllerName) {
|
|
klog.Warningf("%q is disabled", controllerName)
|
|
continue
|
|
}
|
|
|
|
time.Sleep(wait.Jitter(ctx.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))
|
|
|
|
klog.V(1).Infof("Starting %q", controllerName)
|
|
debugHandler, started, err := initFn(ctx)
|
|
if err != nil {
|
|
klog.Errorf("Error starting %q", controllerName)
|
|
return err
|
|
}
|
|
if !started {
|
|
klog.Warningf("Skipping %q", controllerName)
|
|
continue
|
|
}
|
|
if debugHandler != nil && unsecuredMux != nil {
|
|
basePath := "/debug/controllers/" + controllerName
|
|
unsecuredMux.UnlistedHandle(basePath, http.StripPrefix(basePath, debugHandler))
|
|
unsecuredMux.UnlistedHandlePrefix(basePath+"/", http.StripPrefix(basePath, debugHandler))
|
|
}
|
|
klog.Infof("Started %q", controllerName)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// serviceAccountTokenControllerStarter is special because it must run first to set up permissions for other controllers.
|
|
// It cannot use the "normal" client builder, so it tracks its own. It must also avoid being included in the "normal"
|
|
// init map so that it can always run first.
|
|
type serviceAccountTokenControllerStarter struct {
|
|
rootClientBuilder clientbuilder.ControllerClientBuilder
|
|
}
|
|
|
|
func (c serviceAccountTokenControllerStarter) startServiceAccountTokenController(ctx ControllerContext) (http.Handler, bool, error) {
|
|
if !ctx.IsControllerEnabled(saTokenControllerName) {
|
|
klog.Warningf("%q is disabled", saTokenControllerName)
|
|
return nil, false, nil
|
|
}
|
|
|
|
if len(ctx.ComponentConfig.SAController.ServiceAccountKeyFile) == 0 {
|
|
klog.Warningf("%q is disabled because there is no private key", saTokenControllerName)
|
|
return nil, false, nil
|
|
}
|
|
privateKey, err := keyutil.PrivateKeyFromFile(ctx.ComponentConfig.SAController.ServiceAccountKeyFile)
|
|
if err != nil {
|
|
return nil, true, fmt.Errorf("error reading key for service account token controller: %v", err)
|
|
}
|
|
|
|
var rootCA []byte
|
|
if ctx.ComponentConfig.SAController.RootCAFile != "" {
|
|
if rootCA, err = readCA(ctx.ComponentConfig.SAController.RootCAFile); err != nil {
|
|
return nil, true, fmt.Errorf("error parsing root-ca-file at %s: %v", ctx.ComponentConfig.SAController.RootCAFile, err)
|
|
}
|
|
} else {
|
|
rootCA = c.rootClientBuilder.ConfigOrDie("tokens-controller").CAData
|
|
}
|
|
|
|
tokenGenerator, err := serviceaccount.JWTTokenGenerator(serviceaccount.LegacyIssuer, privateKey)
|
|
if err != nil {
|
|
return nil, false, fmt.Errorf("failed to build token generator: %v", err)
|
|
}
|
|
controller, err := serviceaccountcontroller.NewTokensController(
|
|
ctx.InformerFactory.Core().V1().ServiceAccounts(),
|
|
ctx.InformerFactory.Core().V1().Secrets(),
|
|
c.rootClientBuilder.ClientOrDie("tokens-controller"),
|
|
serviceaccountcontroller.TokensControllerOptions{
|
|
TokenGenerator: tokenGenerator,
|
|
RootCA: rootCA,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, true, fmt.Errorf("error creating Tokens controller: %v", err)
|
|
}
|
|
go controller.Run(int(ctx.ComponentConfig.SAController.ConcurrentSATokenSyncs), ctx.Stop)
|
|
|
|
// start the first set of informers now so that other controllers can start
|
|
ctx.InformerFactory.Start(ctx.Stop)
|
|
|
|
return nil, true, nil
|
|
}
|
|
|
|
func readCA(file string) ([]byte, error) {
|
|
rootCA, err := ioutil.ReadFile(file)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err := certutil.ParseCertsPEM(rootCA); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return rootCA, err
|
|
}
|
|
|
|
// createClientBuilders creates clientBuilder and rootClientBuilder from the given configuration
|
|
func createClientBuilders(c *config.CompletedConfig) (clientBuilder clientbuilder.ControllerClientBuilder, rootClientBuilder clientbuilder.ControllerClientBuilder) {
|
|
rootClientBuilder = clientbuilder.SimpleControllerClientBuilder{
|
|
ClientConfig: c.Kubeconfig,
|
|
}
|
|
if c.ComponentConfig.KubeCloudShared.UseServiceAccountCredentials {
|
|
if len(c.ComponentConfig.SAController.ServiceAccountKeyFile) == 0 {
|
|
// It's possible another controller process is creating the tokens for us.
|
|
// If one isn't, we'll timeout and exit when our client builder is unable to create the tokens.
|
|
klog.Warningf("--use-service-account-credentials was specified without providing a --service-account-private-key-file")
|
|
}
|
|
|
|
clientBuilder = clientbuilder.NewDynamicClientBuilder(
|
|
restclient.AnonymousClientConfig(c.Kubeconfig),
|
|
c.Client.CoreV1(),
|
|
metav1.NamespaceSystem)
|
|
} else {
|
|
clientBuilder = rootClientBuilder
|
|
}
|
|
return
|
|
}
|
|
|
|
// leaderElectAndRun runs the leader election, and runs the callbacks once the leader lease is acquired.
|
|
// TODO: extract this function into staging/controller-manager
|
|
func leaderElectAndRun(c *config.CompletedConfig, lockIdentity string, electionChecker *leaderelection.HealthzAdaptor, resourceLock string, leaseName string, callbacks leaderelection.LeaderCallbacks) {
|
|
rl, err := resourcelock.NewFromKubeconfig(resourceLock,
|
|
c.ComponentConfig.Generic.LeaderElection.ResourceNamespace,
|
|
leaseName,
|
|
resourcelock.ResourceLockConfig{
|
|
Identity: lockIdentity,
|
|
EventRecorder: c.EventRecorder,
|
|
},
|
|
c.Kubeconfig,
|
|
c.ComponentConfig.Generic.LeaderElection.RenewDeadline.Duration)
|
|
if err != nil {
|
|
klog.Fatalf("error creating lock: %v", err)
|
|
}
|
|
|
|
leaderelection.RunOrDie(context.TODO(), leaderelection.LeaderElectionConfig{
|
|
Lock: rl,
|
|
LeaseDuration: c.ComponentConfig.Generic.LeaderElection.LeaseDuration.Duration,
|
|
RenewDeadline: c.ComponentConfig.Generic.LeaderElection.RenewDeadline.Duration,
|
|
RetryPeriod: c.ComponentConfig.Generic.LeaderElection.RetryPeriod.Duration,
|
|
Callbacks: callbacks,
|
|
WatchDog: electionChecker,
|
|
Name: leaseName,
|
|
})
|
|
|
|
panic("unreachable")
|
|
}
|
|
|
|
// createInitializersFunc creates a initializersFunc that returns all initializer
|
|
// with expected as the result after filtering through filterFunc.
|
|
func createInitializersFunc(filterFunc leadermigration.FilterFunc, expected leadermigration.FilterResult) ControllerInitializersFunc {
|
|
return func(loopMode ControllerLoopMode) map[string]InitFunc {
|
|
initializers := make(map[string]InitFunc)
|
|
for name, initializer := range NewControllerInitializers(loopMode) {
|
|
if filterFunc(name) == expected {
|
|
initializers[name] = initializer
|
|
}
|
|
}
|
|
return initializers
|
|
}
|
|
}
|