mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-31 15:25:57 +00:00
Merge pull request #46783 from deads2k/controller-08-initfn
Automatic merge from submit-queue (batch tested with PRs 40760, 46706, 46783, 46742, 46751) complete the controller context for init funcs This completes the conversion to initFuncs for the controller initialization to make easier and more manageable to add them.
This commit is contained in:
Kubernetes Submit Queue
GitHub
commit
638c7382ae
@ -37,7 +37,6 @@ import (
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
|
||||
"k8s.io/apiserver/pkg/server/healthz"
|
||||
utilfeature "k8s.io/apiserver/pkg/util/feature"
|
||||
|
||||
"k8s.io/client-go/discovery"
|
||||
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
|
||||
@ -55,13 +54,7 @@ import (
|
||||
"k8s.io/kubernetes/pkg/client/leaderelection/resourcelock"
|
||||
"k8s.io/kubernetes/pkg/cloudprovider"
|
||||
"k8s.io/kubernetes/pkg/controller"
|
||||
nodecontroller "k8s.io/kubernetes/pkg/controller/node"
|
||||
routecontroller "k8s.io/kubernetes/pkg/controller/route"
|
||||
servicecontroller "k8s.io/kubernetes/pkg/controller/service"
|
||||
serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
|
||||
"k8s.io/kubernetes/pkg/controller/volume/attachdetach"
|
||||
persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
|
||||
"k8s.io/kubernetes/pkg/features"
|
||||
"k8s.io/kubernetes/pkg/serviceaccount"
|
||||
"k8s.io/kubernetes/pkg/util/configz"
|
||||
|
||||
@ -175,10 +168,19 @@ func Run(s *options.CMServer) error {
|
||||
} else {
|
||||
clientBuilder = rootClientBuilder
|
||||
}
|
||||
ctx, err := CreateControllerContext(s, rootClientBuilder, clientBuilder, stop)
|
||||
if err != nil {
|
||||
glog.Fatalf("error building controller context: %v", err)
|
||||
}
|
||||
saTokenControllerInitFunc := serviceAccountTokenControllerStarter{rootClientBuilder: rootClientBuilder}.startServiceAccountTokenController
|
||||
|
||||
err := StartControllers(NewControllerInitializers(), s, rootClientBuilder, clientBuilder, stop)
|
||||
glog.Fatalf("error running controllers: %v", err)
|
||||
panic("unreachable")
|
||||
if err := StartControllers(ctx, saTokenControllerInitFunc, NewControllerInitializers()); err != nil {
|
||||
glog.Fatalf("error starting controllers: %v", err)
|
||||
}
|
||||
|
||||
ctx.InformerFactory.Start(ctx.Stop)
|
||||
|
||||
select {}
|
||||
}
|
||||
|
||||
if !s.LeaderElection.LeaderElect {
|
||||
@ -231,6 +233,10 @@ type ControllerContext struct {
|
||||
// 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
|
||||
|
||||
// Stop is the stop channel
|
||||
Stop <-chan struct{}
|
||||
}
|
||||
@ -272,16 +278,14 @@ type InitFunc func(ctx ControllerContext) (bool, error)
|
||||
func KnownControllers() []string {
|
||||
ret := sets.StringKeySet(NewControllerInitializers())
|
||||
|
||||
// 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,
|
||||
nodeControllerName,
|
||||
serviceControllerName,
|
||||
routeControllerName,
|
||||
pvBinderControllerName,
|
||||
attachDetachControllerName,
|
||||
)
|
||||
|
||||
// add "special" controllers that aren't initialized normally
|
||||
return ret.List()
|
||||
}
|
||||
|
||||
@ -290,6 +294,10 @@ var ControllersDisabledByDefault = sets.NewString(
|
||||
"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() map[string]InitFunc {
|
||||
@ -314,6 +322,11 @@ func NewControllerInitializers() map[string]InitFunc {
|
||||
controllers["ttl"] = startTTLController
|
||||
controllers["bootstrapsigner"] = startBootstrapSignerController
|
||||
controllers["tokencleaner"] = startTokenCleanerController
|
||||
controllers["service"] = startServiceController
|
||||
controllers["node"] = startNodeController
|
||||
controllers["route"] = startRouteController
|
||||
controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
|
||||
controllers["attachdetach"] = startAttachDetachController
|
||||
|
||||
return controllers
|
||||
}
|
||||
@ -366,61 +379,22 @@ func GetAvailableResources(clientBuilder controller.ControllerClientBuilder) (ma
|
||||
return allResources, nil
|
||||
}
|
||||
|
||||
const (
|
||||
saTokenControllerName = "serviceaccount-token"
|
||||
nodeControllerName = "node"
|
||||
serviceControllerName = "service"
|
||||
routeControllerName = "route"
|
||||
pvBinderControllerName = "persistentvolume-binder"
|
||||
attachDetachControllerName = "attachdetach"
|
||||
)
|
||||
|
||||
func StartControllers(controllers map[string]InitFunc, s *options.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) error {
|
||||
func CreateControllerContext(s *options.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {
|
||||
versionedClient := rootClientBuilder.ClientOrDie("shared-informers")
|
||||
sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())
|
||||
|
||||
// always start the SA token controller first using a full-power client, since it needs to mint tokens for the rest
|
||||
if len(s.ServiceAccountKeyFile) > 0 && IsControllerEnabled(saTokenControllerName, ControllersDisabledByDefault, s.Controllers...) {
|
||||
privateKey, err := serviceaccount.ReadPrivateKey(s.ServiceAccountKeyFile)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error reading key for service account token controller: %v", err)
|
||||
} else {
|
||||
var rootCA []byte
|
||||
if s.RootCAFile != "" {
|
||||
rootCA, err = ioutil.ReadFile(s.RootCAFile)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error reading root-ca-file at %s: %v", s.RootCAFile, err)
|
||||
}
|
||||
if _, err := certutil.ParseCertsPEM(rootCA); err != nil {
|
||||
return fmt.Errorf("error parsing root-ca-file at %s: %v", s.RootCAFile, err)
|
||||
}
|
||||
} else {
|
||||
rootCA = rootClientBuilder.ConfigOrDie("tokens-controller").CAData
|
||||
}
|
||||
|
||||
controller := serviceaccountcontroller.NewTokensController(
|
||||
sharedInformers.Core().V1().ServiceAccounts(),
|
||||
sharedInformers.Core().V1().Secrets(),
|
||||
rootClientBuilder.ClientOrDie("tokens-controller"),
|
||||
serviceaccountcontroller.TokensControllerOptions{
|
||||
TokenGenerator: serviceaccount.JWTTokenGenerator(privateKey),
|
||||
RootCA: rootCA,
|
||||
},
|
||||
)
|
||||
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
go controller.Run(int(s.ConcurrentSATokenSyncs), stop)
|
||||
|
||||
// start the first set of informers now so that other controllers can start
|
||||
sharedInformers.Start(stop)
|
||||
}
|
||||
|
||||
} else {
|
||||
glog.Warningf("%q is disabled", saTokenControllerName)
|
||||
availableResources, err := GetAvailableResources(rootClientBuilder)
|
||||
if err != nil {
|
||||
return ControllerContext{}, err
|
||||
}
|
||||
|
||||
availableResources, err := GetAvailableResources(clientBuilder)
|
||||
cloud, err := cloudprovider.InitCloudProvider(s.CloudProvider, s.CloudConfigFile)
|
||||
if err != nil {
|
||||
return err
|
||||
return ControllerContext{}, fmt.Errorf("cloud provider could not be initialized: %v", err)
|
||||
}
|
||||
if cloud != nil {
|
||||
// Initialize the cloud provider with a reference to the clientBuilder
|
||||
cloud.Initialize(rootClientBuilder)
|
||||
}
|
||||
|
||||
ctx := ControllerContext{
|
||||
@ -428,8 +402,18 @@ func StartControllers(controllers map[string]InitFunc, s *options.CMServer, root
|
||||
InformerFactory: sharedInformers,
|
||||
Options: *s,
|
||||
AvailableResources: availableResources,
|
||||
Cloud: cloud,
|
||||
Stop: stop,
|
||||
}
|
||||
return ctx, nil
|
||||
}
|
||||
|
||||
func StartControllers(ctx ControllerContext, startSATokenController InitFunc, controllers map[string]InitFunc) 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
|
||||
}
|
||||
|
||||
for controllerName, initFn := range controllers {
|
||||
if !ctx.IsControllerEnabled(controllerName) {
|
||||
@ -437,7 +421,7 @@ func StartControllers(controllers map[string]InitFunc, s *options.CMServer, root
|
||||
continue
|
||||
}
|
||||
|
||||
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
time.Sleep(wait.Jitter(ctx.Options.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
|
||||
glog.V(1).Infof("Starting %q", controllerName)
|
||||
started, err := initFn(ctx)
|
||||
@ -452,144 +436,57 @@ func StartControllers(controllers map[string]InitFunc, s *options.CMServer, root
|
||||
glog.Infof("Started %q", controllerName)
|
||||
}
|
||||
|
||||
// all the remaining plugins want this cloud variable
|
||||
cloud, err := cloudprovider.InitCloudProvider(s.CloudProvider, s.CloudConfigFile)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cloud provider could not be initialized: %v", err)
|
||||
}
|
||||
|
||||
if cloud != nil {
|
||||
// Initialize the cloud provider with a reference to the clientBuilder
|
||||
cloud.Initialize(clientBuilder)
|
||||
}
|
||||
|
||||
if ctx.IsControllerEnabled(nodeControllerName) {
|
||||
_, clusterCIDR, err := net.ParseCIDR(s.ClusterCIDR)
|
||||
if err != nil {
|
||||
glog.Warningf("Unsuccessful parsing of cluster CIDR %v: %v", s.ClusterCIDR, err)
|
||||
}
|
||||
_, serviceCIDR, err := net.ParseCIDR(s.ServiceCIDR)
|
||||
if err != nil {
|
||||
glog.Warningf("Unsuccessful parsing of service CIDR %v: %v", s.ServiceCIDR, err)
|
||||
}
|
||||
nodeController, err := nodecontroller.NewNodeController(
|
||||
sharedInformers.Core().V1().Pods(),
|
||||
sharedInformers.Core().V1().Nodes(),
|
||||
sharedInformers.Extensions().V1beta1().DaemonSets(),
|
||||
cloud,
|
||||
clientBuilder.ClientOrDie("node-controller"),
|
||||
s.PodEvictionTimeout.Duration,
|
||||
s.NodeEvictionRate,
|
||||
s.SecondaryNodeEvictionRate,
|
||||
s.LargeClusterSizeThreshold,
|
||||
s.UnhealthyZoneThreshold,
|
||||
s.NodeMonitorGracePeriod.Duration,
|
||||
s.NodeStartupGracePeriod.Duration,
|
||||
s.NodeMonitorPeriod.Duration,
|
||||
clusterCIDR,
|
||||
serviceCIDR,
|
||||
int(s.NodeCIDRMaskSize),
|
||||
s.AllocateNodeCIDRs,
|
||||
nodecontroller.CIDRAllocatorType(s.CIDRAllocatorType),
|
||||
s.EnableTaintManager,
|
||||
utilfeature.DefaultFeatureGate.Enabled(features.TaintBasedEvictions),
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to initialize nodecontroller: %v", err)
|
||||
}
|
||||
go nodeController.Run(stop)
|
||||
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
} else {
|
||||
glog.Warningf("%q is disabled", nodeControllerName)
|
||||
}
|
||||
|
||||
if ctx.IsControllerEnabled(serviceControllerName) {
|
||||
serviceController, err := servicecontroller.New(
|
||||
cloud,
|
||||
clientBuilder.ClientOrDie("service-controller"),
|
||||
sharedInformers.Core().V1().Services(),
|
||||
sharedInformers.Core().V1().Nodes(),
|
||||
s.ClusterName,
|
||||
)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to start service controller: %v", err)
|
||||
} else {
|
||||
go serviceController.Run(stop, int(s.ConcurrentServiceSyncs))
|
||||
}
|
||||
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
} else {
|
||||
glog.Warningf("%q is disabled", serviceControllerName)
|
||||
}
|
||||
|
||||
if ctx.IsControllerEnabled(routeControllerName) {
|
||||
_, clusterCIDR, err := net.ParseCIDR(s.ClusterCIDR)
|
||||
if err != nil {
|
||||
glog.Warningf("Unsuccessful parsing of cluster CIDR %v: %v", s.ClusterCIDR, err)
|
||||
}
|
||||
if s.AllocateNodeCIDRs && s.ConfigureCloudRoutes {
|
||||
if cloud == nil {
|
||||
glog.Warning("configure-cloud-routes is set, but no cloud provider specified. Will not configure cloud provider routes.")
|
||||
} else if routes, ok := cloud.Routes(); !ok {
|
||||
glog.Warning("configure-cloud-routes is set, but cloud provider does not support routes. Will not configure cloud provider routes.")
|
||||
} else {
|
||||
routeController := routecontroller.New(routes, clientBuilder.ClientOrDie("route-controller"), sharedInformers.Core().V1().Nodes(), s.ClusterName, clusterCIDR)
|
||||
go routeController.Run(stop, s.RouteReconciliationPeriod.Duration)
|
||||
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
}
|
||||
} else {
|
||||
glog.Infof("Will not configure cloud provider routes for allocate-node-cidrs: %v, configure-cloud-routes: %v.", s.AllocateNodeCIDRs, s.ConfigureCloudRoutes)
|
||||
}
|
||||
} else {
|
||||
glog.Warningf("%q is disabled", routeControllerName)
|
||||
}
|
||||
|
||||
if ctx.IsControllerEnabled(pvBinderControllerName) {
|
||||
params := persistentvolumecontroller.ControllerParameters{
|
||||
KubeClient: clientBuilder.ClientOrDie("persistent-volume-binder"),
|
||||
SyncPeriod: s.PVClaimBinderSyncPeriod.Duration,
|
||||
VolumePlugins: ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
|
||||
Cloud: cloud,
|
||||
ClusterName: s.ClusterName,
|
||||
VolumeInformer: sharedInformers.Core().V1().PersistentVolumes(),
|
||||
ClaimInformer: sharedInformers.Core().V1().PersistentVolumeClaims(),
|
||||
ClassInformer: sharedInformers.Storage().V1().StorageClasses(),
|
||||
EnableDynamicProvisioning: s.VolumeConfiguration.EnableDynamicProvisioning,
|
||||
}
|
||||
volumeController, volumeControllerErr := persistentvolumecontroller.NewController(params)
|
||||
if volumeControllerErr != nil {
|
||||
return fmt.Errorf("failed to construct persistentvolume controller: %v", volumeControllerErr)
|
||||
}
|
||||
go volumeController.Run(stop)
|
||||
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
} else {
|
||||
glog.Warningf("%q is disabled", pvBinderControllerName)
|
||||
}
|
||||
|
||||
if ctx.IsControllerEnabled(attachDetachControllerName) {
|
||||
if s.ReconcilerSyncLoopPeriod.Duration < time.Second {
|
||||
return fmt.Errorf("Duration time must be greater than one second as set via command line option reconcile-sync-loop-period.")
|
||||
}
|
||||
attachDetachController, attachDetachControllerErr :=
|
||||
attachdetach.NewAttachDetachController(
|
||||
clientBuilder.ClientOrDie("attachdetach-controller"),
|
||||
sharedInformers.Core().V1().Pods(),
|
||||
sharedInformers.Core().V1().Nodes(),
|
||||
sharedInformers.Core().V1().PersistentVolumeClaims(),
|
||||
sharedInformers.Core().V1().PersistentVolumes(),
|
||||
cloud,
|
||||
ProbeAttachableVolumePlugins(s.VolumeConfiguration),
|
||||
s.DisableAttachDetachReconcilerSync,
|
||||
s.ReconcilerSyncLoopPeriod.Duration)
|
||||
if attachDetachControllerErr != nil {
|
||||
return fmt.Errorf("failed to start attach/detach controller: %v", attachDetachControllerErr)
|
||||
}
|
||||
go attachDetachController.Run(stop)
|
||||
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
|
||||
} else {
|
||||
glog.Warningf("%q is disabled", attachDetachControllerName)
|
||||
}
|
||||
|
||||
sharedInformers.Start(stop)
|
||||
|
||||
select {}
|
||||
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) (bool, error) {
|
||||
if !ctx.IsControllerEnabled(saTokenControllerName) {
|
||||
glog.Warningf("%q is disabled", saTokenControllerName)
|
||||
return false, nil
|
||||
}
|
||||
|
||||
if len(ctx.Options.ServiceAccountKeyFile) == 0 {
|
||||
glog.Warningf("%q is disabled because there is no private key", saTokenControllerName)
|
||||
return false, nil
|
||||
}
|
||||
privateKey, err := serviceaccount.ReadPrivateKey(ctx.Options.ServiceAccountKeyFile)
|
||||
if err != nil {
|
||||
return true, fmt.Errorf("error reading key for service account token controller: %v", err)
|
||||
}
|
||||
|
||||
var rootCA []byte
|
||||
if ctx.Options.RootCAFile != "" {
|
||||
rootCA, err = ioutil.ReadFile(ctx.Options.RootCAFile)
|
||||
if err != nil {
|
||||
return true, fmt.Errorf("error reading root-ca-file at %s: %v", ctx.Options.RootCAFile, err)
|
||||
}
|
||||
if _, err := certutil.ParseCertsPEM(rootCA); err != nil {
|
||||
return true, fmt.Errorf("error parsing root-ca-file at %s: %v", ctx.Options.RootCAFile, err)
|
||||
}
|
||||
} else {
|
||||
rootCA = c.rootClientBuilder.ConfigOrDie("tokens-controller").CAData
|
||||
}
|
||||
|
||||
controller := serviceaccountcontroller.NewTokensController(
|
||||
ctx.InformerFactory.Core().V1().ServiceAccounts(),
|
||||
ctx.InformerFactory.Core().V1().Secrets(),
|
||||
c.rootClientBuilder.ClientOrDie("tokens-controller"),
|
||||
serviceaccountcontroller.TokensControllerOptions{
|
||||
TokenGenerator: serviceaccount.JWTTokenGenerator(privateKey),
|
||||
RootCA: rootCA,
|
||||
},
|
||||
)
|
||||
go controller.Run(int(ctx.Options.ConcurrentSATokenSyncs), ctx.Stop)
|
||||
|
||||
// start the first set of informers now so that other controllers can start
|
||||
ctx.InformerFactory.Start(ctx.Stop)
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
@ -22,10 +22,15 @@ package app
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/golang/glog"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
utilfeature "k8s.io/apiserver/pkg/util/feature"
|
||||
"k8s.io/client-go/discovery"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
@ -37,14 +42,140 @@ import (
|
||||
"k8s.io/kubernetes/pkg/controller/garbagecollector"
|
||||
"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
|
||||
namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace"
|
||||
nodecontroller "k8s.io/kubernetes/pkg/controller/node"
|
||||
"k8s.io/kubernetes/pkg/controller/podgc"
|
||||
replicationcontroller "k8s.io/kubernetes/pkg/controller/replication"
|
||||
resourcequotacontroller "k8s.io/kubernetes/pkg/controller/resourcequota"
|
||||
routecontroller "k8s.io/kubernetes/pkg/controller/route"
|
||||
servicecontroller "k8s.io/kubernetes/pkg/controller/service"
|
||||
serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
|
||||
ttlcontroller "k8s.io/kubernetes/pkg/controller/ttl"
|
||||
"k8s.io/kubernetes/pkg/controller/volume/attachdetach"
|
||||
persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
|
||||
"k8s.io/kubernetes/pkg/features"
|
||||
quotainstall "k8s.io/kubernetes/pkg/quota/install"
|
||||
)
|
||||
|
||||
func startServiceController(ctx ControllerContext) (bool, error) {
|
||||
serviceController, err := servicecontroller.New(
|
||||
ctx.Cloud,
|
||||
ctx.ClientBuilder.ClientOrDie("service-controller"),
|
||||
ctx.InformerFactory.Core().V1().Services(),
|
||||
ctx.InformerFactory.Core().V1().Nodes(),
|
||||
ctx.Options.ClusterName,
|
||||
)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to start service controller: %v", err)
|
||||
return false, nil
|
||||
}
|
||||
go serviceController.Run(ctx.Stop, int(ctx.Options.ConcurrentServiceSyncs))
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func startNodeController(ctx ControllerContext) (bool, error) {
|
||||
_, clusterCIDR, err := net.ParseCIDR(ctx.Options.ClusterCIDR)
|
||||
if err != nil {
|
||||
glog.Warningf("Unsuccessful parsing of cluster CIDR %v: %v", ctx.Options.ClusterCIDR, err)
|
||||
}
|
||||
_, serviceCIDR, err := net.ParseCIDR(ctx.Options.ServiceCIDR)
|
||||
if err != nil {
|
||||
glog.Warningf("Unsuccessful parsing of service CIDR %v: %v", ctx.Options.ServiceCIDR, err)
|
||||
}
|
||||
nodeController, err := nodecontroller.NewNodeController(
|
||||
ctx.InformerFactory.Core().V1().Pods(),
|
||||
ctx.InformerFactory.Core().V1().Nodes(),
|
||||
ctx.InformerFactory.Extensions().V1beta1().DaemonSets(),
|
||||
ctx.Cloud,
|
||||
ctx.ClientBuilder.ClientOrDie("node-controller"),
|
||||
ctx.Options.PodEvictionTimeout.Duration,
|
||||
ctx.Options.NodeEvictionRate,
|
||||
ctx.Options.SecondaryNodeEvictionRate,
|
||||
ctx.Options.LargeClusterSizeThreshold,
|
||||
ctx.Options.UnhealthyZoneThreshold,
|
||||
ctx.Options.NodeMonitorGracePeriod.Duration,
|
||||
ctx.Options.NodeStartupGracePeriod.Duration,
|
||||
ctx.Options.NodeMonitorPeriod.Duration,
|
||||
clusterCIDR,
|
||||
serviceCIDR,
|
||||
int(ctx.Options.NodeCIDRMaskSize),
|
||||
ctx.Options.AllocateNodeCIDRs,
|
||||
nodecontroller.CIDRAllocatorType(ctx.Options.CIDRAllocatorType),
|
||||
ctx.Options.EnableTaintManager,
|
||||
utilfeature.DefaultFeatureGate.Enabled(features.TaintBasedEvictions),
|
||||
)
|
||||
if err != nil {
|
||||
return true, err
|
||||
}
|
||||
go nodeController.Run(ctx.Stop)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func startRouteController(ctx ControllerContext) (bool, error) {
|
||||
_, clusterCIDR, err := net.ParseCIDR(ctx.Options.ClusterCIDR)
|
||||
if err != nil {
|
||||
glog.Warningf("Unsuccessful parsing of cluster CIDR %v: %v", ctx.Options.ClusterCIDR, err)
|
||||
}
|
||||
// TODO demorgans
|
||||
if ctx.Options.AllocateNodeCIDRs && ctx.Options.ConfigureCloudRoutes {
|
||||
if ctx.Cloud == nil {
|
||||
glog.Warning("configure-cloud-routes is set, but no cloud provider specified. Will not configure cloud provider routes.")
|
||||
return false, nil
|
||||
} else if routes, ok := ctx.Cloud.Routes(); !ok {
|
||||
glog.Warning("configure-cloud-routes is set, but cloud provider does not support routes. Will not configure cloud provider routes.")
|
||||
return false, nil
|
||||
} else {
|
||||
routeController := routecontroller.New(routes, ctx.ClientBuilder.ClientOrDie("route-controller"), ctx.InformerFactory.Core().V1().Nodes(), ctx.Options.ClusterName, clusterCIDR)
|
||||
go routeController.Run(ctx.Stop, ctx.Options.RouteReconciliationPeriod.Duration)
|
||||
return true, nil
|
||||
}
|
||||
} else {
|
||||
glog.Infof("Will not configure cloud provider routes for allocate-node-cidrs: %v, configure-cloud-routes: %v.", ctx.Options.AllocateNodeCIDRs, ctx.Options.ConfigureCloudRoutes)
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
|
||||
func startPersistentVolumeBinderController(ctx ControllerContext) (bool, error) {
|
||||
params := persistentvolumecontroller.ControllerParameters{
|
||||
KubeClient: ctx.ClientBuilder.ClientOrDie("persistent-volume-binder"),
|
||||
SyncPeriod: ctx.Options.PVClaimBinderSyncPeriod.Duration,
|
||||
VolumePlugins: ProbeControllerVolumePlugins(ctx.Cloud, ctx.Options.VolumeConfiguration),
|
||||
Cloud: ctx.Cloud,
|
||||
ClusterName: ctx.Options.ClusterName,
|
||||
VolumeInformer: ctx.InformerFactory.Core().V1().PersistentVolumes(),
|
||||
ClaimInformer: ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
|
||||
ClassInformer: ctx.InformerFactory.Storage().V1().StorageClasses(),
|
||||
EnableDynamicProvisioning: ctx.Options.VolumeConfiguration.EnableDynamicProvisioning,
|
||||
}
|
||||
volumeController, volumeControllerErr := persistentvolumecontroller.NewController(params)
|
||||
if volumeControllerErr != nil {
|
||||
return true, fmt.Errorf("failed to construct persistentvolume controller: %v", volumeControllerErr)
|
||||
}
|
||||
go volumeController.Run(ctx.Stop)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func startAttachDetachController(ctx ControllerContext) (bool, error) {
|
||||
if ctx.Options.ReconcilerSyncLoopPeriod.Duration < time.Second {
|
||||
return true, fmt.Errorf("Duration time must be greater than one second as set via command line option reconcile-sync-loop-period.")
|
||||
}
|
||||
attachDetachController, attachDetachControllerErr :=
|
||||
attachdetach.NewAttachDetachController(
|
||||
ctx.ClientBuilder.ClientOrDie("attachdetach-controller"),
|
||||
ctx.InformerFactory.Core().V1().Pods(),
|
||||
ctx.InformerFactory.Core().V1().Nodes(),
|
||||
ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
|
||||
ctx.InformerFactory.Core().V1().PersistentVolumes(),
|
||||
ctx.Cloud,
|
||||
ProbeAttachableVolumePlugins(ctx.Options.VolumeConfiguration),
|
||||
ctx.Options.DisableAttachDetachReconcilerSync,
|
||||
ctx.Options.ReconcilerSyncLoopPeriod.Duration)
|
||||
if attachDetachControllerErr != nil {
|
||||
return true, fmt.Errorf("failed to start attach/detach controller: %v", attachDetachControllerErr)
|
||||
}
|
||||
go attachDetachController.Run(ctx.Stop)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func startEndpointController(ctx ControllerContext) (bool, error) {
|
||||
go endpointcontroller.NewEndpointController(
|
||||
ctx.InformerFactory.Core().V1().Pods(),
|
||||
|
||||
Loading…
Reference in New Issue
Block a user