mirror of https://github.com/k3s-io/k3s
626 lines
25 KiB
Go
626 lines
25 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"
|
|
|
|
v1 "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/informers"
|
|
clientset "k8s.io/client-go/kubernetes"
|
|
"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/version"
|
|
"k8s.io/component-base/version/verflag"
|
|
"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"
|
|
)
|
|
|
|
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())
|
|
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
|
|
}
|
|
}
|
|
|
|
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.ObjectOrMetadataInformerFactory.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,
|
|
c.ComponentConfig.Generic.LeaderElection.ResourceNamespace,
|
|
c.ComponentConfig.Generic.LeaderElection.ResourceName,
|
|
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
|
|
|
|
// 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 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["endpointslice"] = startEndpointSliceController
|
|
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.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 controller.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.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,
|
|
ObjectOrMetadataInformerFactory: controller.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 _, _, 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
|
|
}
|