123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625 |
- /*
- 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"
- "k8s.io/api/core/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"
- "k8s.io/apiserver/pkg/server"
- "k8s.io/apiserver/pkg/server/healthz"
- "k8s.io/apiserver/pkg/server/mux"
- utilfeature "k8s.io/apiserver/pkg/util/feature"
- "k8s.io/apiserver/pkg/util/term"
- cacheddiscovery "k8s.io/client-go/discovery/cached"
- "k8s.io/client-go/dynamic"
- "k8s.io/client-go/dynamic/dynamicinformer"
- "k8s.io/client-go/informers"
- clientset "k8s.io/client-go/kubernetes"
- 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/klog"
- genericcontrollermanager "k8s.io/kubernetes/cmd/controller-manager/app"
- "k8s.io/kubernetes/cmd/kube-controller-manager/app/config"
- "k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
- "k8s.io/kubernetes/pkg/controller"
- kubectrlmgrconfig "k8s.io/kubernetes/pkg/controller/apis/config"
- serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
- "k8s.io/kubernetes/pkg/features"
- "k8s.io/kubernetes/pkg/serviceaccount"
- "k8s.io/kubernetes/pkg/util/configz"
- utilflag "k8s.io/kubernetes/pkg/util/flag"
- "k8s.io/kubernetes/pkg/version"
- "k8s.io/kubernetes/pkg/version/verflag"
- )
- 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.`,
- Run: func(cmd *cobra.Command, args []string) {
- verflag.PrintAndExitIfRequested()
- utilflag.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)
- }
- },
- }
- fs := cmd.Flags()
- namedFlagSets := s.Flags(KnownControllers(), ControllersDisabledByDefault.List())
- verflag.AddFlags(namedFlagSets.FlagSet("global"))
- globalflag.AddGlobalFlags(namedFlagSets.FlagSet("global"), cmd.Name())
- // hoist this flag from the global flagset to preserve the commandline until
- // the gce cloudprovider is removed.
- globalflag.Register(namedFlagSets.FlagSet("generic"), "cloud-provider-gce-lb-src-cidrs")
- namedFlagSets.FlagSet("generic").MarkDeprecated("cloud-provider-gce-lb-src-cidrs", "This flag will be removed once the GCE Cloud Provider is removed from kube-controller-manager")
- 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: %c", err)
- }
- // Setup any healthz checks we will want to use.
- var checks []healthz.HealthzChecker
- 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
- }
- }
- run := func(ctx context.Context) {
- rootClientBuilder := controller.SimpleControllerClientBuilder{
- ClientConfig: c.Kubeconfig,
- }
- var clientBuilder controller.ControllerClientBuilder
- 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")
- }
- if shouldTurnOnDynamicClient(c.Client) {
- klog.V(1).Infof("using dynamic client builder")
- //Dynamic builder will use TokenRequest feature and refresh service account token periodically
- clientBuilder = controller.NewDynamicClientBuilder(
- restclient.AnonymousClientConfig(c.Kubeconfig),
- c.Client.CoreV1(),
- "kube-system")
- } else {
- klog.V(1).Infof("using legacy client builder")
- clientBuilder = controller.SAControllerClientBuilder{
- ClientConfig: restclient.AnonymousClientConfig(c.Kubeconfig),
- CoreClient: c.Client.CoreV1(),
- AuthenticationClient: c.Client.AuthenticationV1(),
- Namespace: "kube-system",
- }
- }
- } else {
- clientBuilder = rootClientBuilder
- }
- controllerContext, err := CreateControllerContext(c, rootClientBuilder, clientBuilder, ctx.Done())
- if err != nil {
- klog.Fatalf("error building controller context: %v", err)
- }
- saTokenControllerInitFunc := serviceAccountTokenControllerStarter{rootClientBuilder: rootClientBuilder}.startServiceAccountTokenController
- if err := StartControllers(controllerContext, saTokenControllerInitFunc, NewControllerInitializers(controllerContext.LoopMode), unsecuredMux); err != nil {
- klog.Fatalf("error starting controllers: %v", err)
- }
- controllerContext.InformerFactory.Start(controllerContext.Stop)
- controllerContext.GenericInformerFactory.Start(controllerContext.Stop)
- close(controllerContext.InformersStarted)
- select {}
- }
- if !c.ComponentConfig.Generic.LeaderElection.LeaderElect {
- run(context.TODO())
- 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())
- rl, err := resourcelock.New(c.ComponentConfig.Generic.LeaderElection.ResourceLock,
- "kube-system",
- "kube-controller-manager",
- c.LeaderElectionClient.CoreV1(),
- c.LeaderElectionClient.CoordinationV1(),
- resourcelock.ResourceLockConfig{
- Identity: id,
- EventRecorder: c.EventRecorder,
- })
- 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: leaderelection.LeaderCallbacks{
- OnStartedLeading: run,
- OnStoppedLeading: func() {
- klog.Fatalf("leaderelection lost")
- },
- },
- WatchDog: electionChecker,
- Name: "kube-controller-manager",
- })
- panic("unreachable")
- }
- // ControllerContext defines the context object for controller
- type ControllerContext struct {
- // ClientBuilder will provide a client for this controller to use
- ClientBuilder controller.ControllerClientBuilder
- // InformerFactory gives access to informers for the controller.
- InformerFactory informers.SharedInformerFactory
- // GenericInformerFactory gives access to informers for typed resources
- // and dynamic resources.
- GenericInformerFactory controller.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)
- // 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["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
- 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 controller.ControllerClientBuilder) (map[schema.GroupVersionResource]bool, error) {
- client := clientBuilder.ClientOrDie("controller-discovery")
- discoveryClient := client.Discovery()
- resourceMap, err := discoveryClient.ServerResources()
- 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 controller.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {
- versionedClient := rootClientBuilder.ClientOrDie("shared-informers")
- sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())
- dynamicClient := dynamic.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("dynamic-informers"))
- dynamicInformers := dynamicinformer.NewDynamicSharedInformerFactory(dynamicClient, 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, 10*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.ClientOrDie("controller-discovery")
- cachedClient := cacheddiscovery.NewMemCacheClient(discoveryClient.Discovery())
- 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,
- GenericInformerFactory: controller.NewInformerFactory(sharedInformers, dynamicInformers),
- 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 _, _, 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 controller.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
- }
- func shouldTurnOnDynamicClient(client clientset.Interface) bool {
- if !utilfeature.DefaultFeatureGate.Enabled(features.TokenRequest) {
- return false
- }
- apiResourceList, err := client.Discovery().ServerResourcesForGroupVersion(v1.SchemeGroupVersion.String())
- if err != nil {
- klog.Warningf("fetch api resource lists failed, use legacy client builder: %v", err)
- return false
- }
- for _, resource := range apiResourceList.APIResources {
- if resource.Name == "serviceaccounts/token" &&
- resource.Group == "authentication.k8s.io" &&
- sets.NewString(resource.Verbs...).Has("create") {
- return true
- }
- }
- return false
- }
|