// Cluster holds a variety of methods to interact with a cluster.

cluster.Cluster

// Add will set requested dependencies on the component, and cause the component to be

// started when Start is called. Add will inject any dependencies for which the argument

// implements the inject interface - e.g. inject.Client.

// Depending on if a Runnable implements LeaderElectionRunnable interface, a Runnable can be run in either

// non-leaderelection mode (always running) or leader election mode (managed by leader election if enabled).

Add(Runnable) error

// Elected is closed when this manager is elected leader of a group of

// managers, either because it won a leader election or because no leader

// election was configured.

Elected() <-chan struct{}

// AddMetricsExtraHandler adds an extra handler served on path to the http server that serves metrics.

// Might be useful to register some diagnostic endpoints e.g. pprof. Note that these endpoints meant to be

// sensitive and shouldn't be exposed publicly.

// If the simple path -> handler mapping offered here is not enough, a new http server/listener should be added as

// Runnable to the manager via Add method.

AddMetricsExtraHandler(path string, handler http.Handler) error

// AddHealthzCheck allows you to add Healthz checker

AddHealthzCheck(name string, check healthz.Checker) error

// AddReadyzCheck allows you to add Readyz checker

AddReadyzCheck(name string, check healthz.Checker) error

// Start starts all registered Controllers and blocks until the context is cancelled.

// Returns an error if there is an error starting any controller.

//

// If LeaderElection is used, the binary must be exited immediately after this returns,

// otherwise components that need leader election might continue to run after the leader

// lock was lost.

Start(ctx context.Context) error

// GetWebhookServer returns a webhook.Server

GetWebhookServer() *webhook.Server

// GetLogger returns this manager's logger.

GetLogger() logr.Logger

// GetControllerOptions returns controller global configuration options.

GetControllerOptions() v1alpha1.ControllerConfigurationSpec

// Scheme is the scheme used to resolve runtime.Objects to GroupVersionKinds / Resources

// Defaults to the kubernetes/client-go scheme.Scheme, but it's almost always better

// idea to pass your own scheme in. See the documentation in pkg/scheme for more information.

Scheme *runtime.Scheme

// MapperProvider provides the rest mapper used to map go types to Kubernetes APIs

MapperProvider func(c *rest.Config) (meta.RESTMapper, error)

// SyncPeriod determines the minimum frequency at which watched resources are

// reconciled. A lower period will correct entropy more quickly, but reduce

// responsiveness to change if there are many watched resources. Change this

// value only if you know what you are doing. Defaults to 10 hours if unset.

// there will a 10 percent jitter between the SyncPeriod of all controllers

// so that all controllers will not send list requests simultaneously.

//

// This applies to all controllers.

//

// A period sync happens for two reasons:

// 1. To insure against a bug in the controller that causes an object to not

// be requeued, when it otherwise should be requeued.

// 2. To insure against an unknown bug in controller-runtime, or its dependencies,

// that causes an object to not be requeued, when it otherwise should be

// requeued, or to be removed from the queue, when it otherwise should not

// be removed.

//

// If you want

// 1. to insure against missed watch events, or

// 2. to poll services that cannot be watched,

// then we recommend that, instead of changing the default period, the

// controller requeue, with a constant duration `t`, whenever the controller

// is "done" with an object, and would otherwise not requeue it, i.e., we

// recommend the `Reconcile` function return `reconcile.Result{RequeueAfter: t}`,

// instead of `reconcile.Result{}`.

SyncPeriod *time.Duration

// Logger is the logger that should be used by this manager.

// If none is set, it defaults to log.Log global logger.

Logger logr.Logger

// LeaderElection determines whether or not to use leader election when

// starting the manager.

LeaderElection bool

// LeaderElectionResourceLock determines which resource lock to use for leader election,

// defaults to "configmapsleases". Change this value only if you know what you are doing.

// Otherwise, users of your controller might end up with multiple running instances that

// each acquired leadership through different resource locks during upgrades and thus

// act on the same resources concurrently.

// If you want to migrate to the "leases" resource lock, you might do so by migrating to the

// respective multilock first ("configmapsleases" or "endpointsleases"), which will acquire a

// leader lock on both resources. After all your users have migrated to the multilock, you can

// go ahead and migrate to "leases". Please also keep in mind, that users might skip versions

// of your controller.

//

// Note: before controller-runtime version v0.7, the resource lock was set to "configmaps".

// Please keep this in mind, when planning a proper migration path for your controller.

LeaderElectionResourceLock string

// LeaderElectionNamespace determines the namespace in which the leader

// election resource will be created.

LeaderElectionNamespace string

// LeaderElectionID determines the name of the resource that leader election

// will use for holding the leader lock.

LeaderElectionID string

// LeaderElectionConfig can be specified to override the default configuration

// that is used to build the leader election client.

LeaderElectionConfig *rest.Config

// LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily

// when the Manager ends. This requires the binary to immediately end when the

// Manager is stopped, otherwise this setting is unsafe. Setting this significantly

// speeds up voluntary leader transitions as the new leader doesn't have to wait

// LeaseDuration time first.

LeaderElectionReleaseOnCancel bool

// LeaseDuration is the duration that non-leader candidates will

// wait to force acquire leadership. This is measured against time of

// last observed ack. Default is 15 seconds.

LeaseDuration *time.Duration

// RenewDeadline is the duration that the acting controlplane will retry

// refreshing leadership before giving up. Default is 10 seconds.

RenewDeadline *time.Duration

// RetryPeriod is the duration the LeaderElector clients should wait

// between tries of actions. Default is 2 seconds.

RetryPeriod *time.Duration

// Namespace if specified restricts the manager's cache to watch objects in

// the desired namespace Defaults to all namespaces

//

// Note: If a namespace is specified, controllers can still Watch for a

// cluster-scoped resource (e.g Node). For namespaced resources the cache

// will only hold objects from the desired namespace.

Namespace string

// MetricsBindAddress is the TCP address that the controller should bind to

// for serving prometheus metrics.

// It can be set to "0" to disable the metrics serving.

MetricsBindAddress string

// HealthProbeBindAddress is the TCP address that the controller should bind to

// for serving health probes

HealthProbeBindAddress string

// Readiness probe endpoint name, defaults to "readyz"

ReadinessEndpointName string

// Liveness probe endpoint name, defaults to "healthz"

LivenessEndpointName string

// Port is the port that the webhook server serves at.

// It is used to set webhook.Server.Port if WebhookServer is not set.

Port int

// Host is the hostname that the webhook server binds to.

// It is used to set webhook.Server.Host if WebhookServer is not set.

Host string

// CertDir is the directory that contains the server key and certificate.

// If not set, webhook server would look up the server key and certificate in

// {TempDir}/k8s-webhook-server/serving-certs. The server key and certificate

// must be named tls.key and tls.crt, respectively.

// It is used to set webhook.Server.CertDir if WebhookServer is not set.

CertDir string

// WebhookServer is an externally configured webhook.Server. By default,

// a Manager will create a default server using Port, Host, and CertDir;

// if this is set, the Manager will use this server instead.

WebhookServer *webhook.Server

// Functions to all for a user to customize the values that will be injected.

// NewCache is the function that will create the cache to be used

// by the manager. If not set this will use the default new cache function.

NewCache cache.NewCacheFunc

// NewClient is the func that creates the client to be used by the manager.

// If not set this will create the default DelegatingClient that will

// use the cache for reads and the client for writes.

NewClient cluster.NewClientFunc

// ClientDisableCacheFor tells the client that, if any cache is used, to bypass it

// for the given objects.

ClientDisableCacheFor []client.Object

// DryRunClient specifies whether the client should be configured to enforce

// dryRun mode.

DryRunClient bool

// EventBroadcaster records Events emitted by the manager and sends them to the Kubernetes API

// Use this to customize the event correlator and spam filter

//

// Deprecated: using this may cause goroutine leaks if the lifetime of your manager or controllers

// is shorter than the lifetime of your process.

EventBroadcaster record.EventBroadcaster

// GracefulShutdownTimeout is the duration given to runnable to stop before the manager actually returns on stop.

// To disable graceful shutdown, set to time.Duration(0)

// To use graceful shutdown without timeout, set to a negative duration, e.G. time.Duration(-1)

// The graceful shutdown is skipped for safety reasons in case the leader election lease is lost.

GracefulShutdownTimeout *time.Duration

// Controller contains global configuration options for controllers

// registered within this manager.

// +optional

Controller v1alpha1.ControllerConfigurationSpec

// makeBroadcaster allows deferring the creation of the broadcaster to

// avoid leaking goroutines if we never call Start on this manager. It also

// returns whether or not this is a "owned" broadcaster, and as such should be

// stopped with the manager.

makeBroadcaster intrec.EventBroadcasterProducer

// Dependency injection for testing

newRecorderProvider func(config *rest.Config, scheme *runtime.Scheme, logger logr.Logger, makeBroadcaster intrec.EventBroadcasterProducer) (*intrec.Provider, error)

newResourceLock func(config *rest.Config, recorderProvider recorder.Provider, options leaderelection.Options) (resourcelock.Interface, error)

newMetricsListener func(addr string) (net.Listener, error)

newHealthProbeListener func(addr string) (net.Listener, error)

// Start starts running the component. The component will stop running

// when the context is closed. Start blocks until the context is closed or

// an error occurs.

Start(context.Context) error

// NeedLeaderElection returns true if the Runnable needs to be run in the leader election mode.

// e.g. controllers need to be run in leader election mode, while webhook server doesn't.

NeedLeaderElection() bool

// Set default values for options fields

options = setOptionsDefaults(options)

cluster, err := cluster.New(config, func(clusterOptions *cluster.Options) {

clusterOptions.Scheme = options.Scheme

clusterOptions.MapperProvider = options.MapperProvider

clusterOptions.Logger = options.Logger

clusterOptions.SyncPeriod = options.SyncPeriod

clusterOptions.Namespace = options.Namespace

clusterOptions.NewCache = options.NewCache

clusterOptions.NewClient = options.NewClient

clusterOptions.ClientDisableCacheFor = options.ClientDisableCacheFor

clusterOptions.DryRunClient = options.DryRunClient

clusterOptions.EventBroadcaster = options.EventBroadcaster

})

if err != nil {

return nil, err

}

// Create the recorder provider to inject event recorders for the components.

// TODO(directxman12): the log for the event provider should have a context (name, tags, etc) specific

// to the particular controller that it's being injected into, rather than a generic one like is here.

recorderProvider, err := options.newRecorderProvider(config, cluster.GetScheme(), options.Logger.WithName("events"), options.makeBroadcaster)

if err != nil {

return nil, err

}

// Create the resource lock to enable leader election)

leaderConfig := options.LeaderElectionConfig

if leaderConfig == nil {

leaderConfig = rest.CopyConfig(config)

}

resourceLock, err := options.newResourceLock(leaderConfig, recorderProvider, leaderelection.Options{

LeaderElection: options.LeaderElection,

LeaderElectionResourceLock: options.LeaderElectionResourceLock,

LeaderElectionID: options.LeaderElectionID,

LeaderElectionNamespace: options.LeaderElectionNamespace,

})

if err != nil {

return nil, err

}

// Create the metrics listener. This will throw an error if the metrics bind

// address is invalid or already in use.

metricsListener, err := options.newMetricsListener(options.MetricsBindAddress)

if err != nil {

return nil, err

}

// By default we have no extra endpoints to expose on metrics http server.

metricsExtraHandlers := make(map[string]http.Handler)

// Create health probes listener. This will throw an error if the bind

// address is invalid or already in use.

healthProbeListener, err := options.newHealthProbeListener(options.HealthProbeBindAddress)

if err != nil {

return nil, err

}

return &controllerManager{

cluster: cluster,

recorderProvider: recorderProvider,

resourceLock: resourceLock,

metricsListener: metricsListener,

metricsExtraHandlers: metricsExtraHandlers,

controllerOptions: options.Controller,

logger: options.Logger,

elected: make(chan struct{}),

port: options.Port,

host: options.Host,

certDir: options.CertDir,

webhookServer: options.WebhookServer,

leaseDuration: *options.LeaseDuration,

renewDeadline: *options.RenewDeadline,

retryPeriod: *options.RetryPeriod,

healthProbeListener: healthProbeListener,

readinessEndpointName: options.ReadinessEndpointName,

livenessEndpointName: options.LivenessEndpointName,

gracefulShutdownTimeout: *options.GracefulShutdownTimeout,

internalProceduresStop: make(chan struct{}),

leaderElectionStopped: make(chan struct{}),

leaderElectionReleaseOnCancel: options.LeaderElectionReleaseOnCancel,

}, nil

if inj, wantsScheme := loader.(inject.Scheme); wantsScheme {

err := inj.InjectScheme(o.Scheme)

if err != nil {

return o, err

}

}

newObj, err := loader.Complete()

if err != nil {

return o, err

}

o = o.setLeaderElectionConfig(newObj)

if o.SyncPeriod == nil && newObj.SyncPeriod != nil {

o.SyncPeriod = &newObj.SyncPeriod.Duration

}

if o.Namespace == "" && newObj.CacheNamespace != "" {

o.Namespace = newObj.CacheNamespace

}

if o.MetricsBindAddress == "" && newObj.Metrics.BindAddress != "" {

o.MetricsBindAddress = newObj.Metrics.BindAddress

}

if o.HealthProbeBindAddress == "" && newObj.Health.HealthProbeBindAddress != "" {

o.HealthProbeBindAddress = newObj.Health.HealthProbeBindAddress

}

if o.ReadinessEndpointName == "" && newObj.Health.ReadinessEndpointName != "" {

o.ReadinessEndpointName = newObj.Health.ReadinessEndpointName

}

if o.LivenessEndpointName == "" && newObj.Health.LivenessEndpointName != "" {

o.LivenessEndpointName = newObj.Health.LivenessEndpointName

}

if o.Port == 0 && newObj.Webhook.Port != nil {

o.Port = *newObj.Webhook.Port

}

if o.Host == "" && newObj.Webhook.Host != "" {

o.Host = newObj.Webhook.Host

}

if o.CertDir == "" && newObj.Webhook.CertDir != "" {

o.CertDir = newObj.Webhook.CertDir

}

if newObj.Controller != nil {

if o.Controller.CacheSyncTimeout == nil && newObj.Controller.CacheSyncTimeout != nil {

o.Controller.CacheSyncTimeout = newObj.Controller.CacheSyncTimeout

}

if len(o.Controller.GroupKindConcurrency) == 0 && len(newObj.Controller.GroupKindConcurrency) > 0 {

o.Controller.GroupKindConcurrency = newObj.Controller.GroupKindConcurrency

}

}

return o, nil

o, err := o.AndFrom(loader)

if err != nil {

panic(fmt.Sprintf("could not parse config file: %v", err))

}

return o

if o.LeaderElection == false && obj.LeaderElection.LeaderElect != nil {

o.LeaderElection = *obj.LeaderElection.LeaderElect

}

if o.LeaderElectionResourceLock == "" && obj.LeaderElection.ResourceLock != "" {

o.LeaderElectionResourceLock = obj.LeaderElection.ResourceLock

}

if o.LeaderElectionNamespace == "" && obj.LeaderElection.ResourceNamespace != "" {

o.LeaderElectionNamespace = obj.LeaderElection.ResourceNamespace

}

if o.LeaderElectionID == "" && obj.LeaderElection.ResourceName != "" {

o.LeaderElectionID = obj.LeaderElection.ResourceName

}

if o.LeaseDuration == nil && !reflect.DeepEqual(obj.LeaderElection.LeaseDuration, metav1.Duration{}) {

o.LeaseDuration = &obj.LeaderElection.LeaseDuration.Duration

}

if o.RenewDeadline == nil && !reflect.DeepEqual(obj.LeaderElection.RenewDeadline, metav1.Duration{}) {

o.RenewDeadline = &obj.LeaderElection.RenewDeadline.Duration

}

if o.RetryPeriod == nil && !reflect.DeepEqual(obj.LeaderElection.RetryPeriod, metav1.Duration{}) {

o.RetryPeriod = &obj.LeaderElection.RetryPeriod.Duration

}

return o

if addr == "" || addr == "0" {

return nil, nil

}

ln, err := net.Listen("tcp", addr)

if err != nil {

return nil, fmt.Errorf("error listening on %s: %v", addr, err)

}

return ln, nil

// Allow newResourceLock to be mocked

if options.newResourceLock == nil {

options.newResourceLock = leaderelection.NewResourceLock

}

// Allow newRecorderProvider to be mocked

if options.newRecorderProvider == nil {

options.newRecorderProvider = intrec.NewProvider

}

// This is duplicated with pkg/cluster, we need it here

// for the leader election and there to provide the user with

// an EventBroadcaster

if options.EventBroadcaster == nil {

// defer initialization to avoid leaking by default

options.makeBroadcaster = func() (record.EventBroadcaster, bool) {

return record.NewBroadcaster(), true

}

} else {

options.makeBroadcaster = func() (record.EventBroadcaster, bool) {

return options.EventBroadcaster, false

}

}

if options.newMetricsListener == nil {

options.newMetricsListener = metrics.NewListener

}

leaseDuration, renewDeadline, retryPeriod := defaultLeaseDuration, defaultRenewDeadline, defaultRetryPeriod

if options.LeaseDuration == nil {

options.LeaseDuration = &leaseDuration

}

if options.RenewDeadline == nil {

options.RenewDeadline = &renewDeadline

}

if options.RetryPeriod == nil {

options.RetryPeriod = &retryPeriod

}

if options.ReadinessEndpointName == "" {

options.ReadinessEndpointName = defaultReadinessEndpoint

}

if options.LivenessEndpointName == "" {

options.LivenessEndpointName = defaultLivenessEndpoint

}

if options.newHealthProbeListener == nil {

options.newHealthProbeListener = defaultHealthProbeListener

}

if options.GracefulShutdownTimeout == nil {

gracefulShutdownTimeout := defaultGracefulShutdownPeriod

options.GracefulShutdownTimeout = &gracefulShutdownTimeout

}

if options.Logger == nil {

options.Logger = logf.RuntimeLog.WithName("manager")

}

return options