forked from openshift/cluster-kube-apiserver-operator
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request openshift#635 from tnozicka/recovery-controller
Add cert regeneration controller
- Loading branch information
Showing
21 changed files
with
778 additions
and
169 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,145 @@ | ||
| package certregenerationcontroller | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| utilruntime "k8s.io/apimachinery/pkg/util/runtime" | ||
| "k8s.io/apimachinery/pkg/util/wait" | ||
| corev1client "k8s.io/client-go/kubernetes/typed/core/v1" | ||
| corev1listers "k8s.io/client-go/listers/core/v1" | ||
| "k8s.io/client-go/tools/cache" | ||
| "k8s.io/client-go/util/workqueue" | ||
| "k8s.io/klog" | ||
|
|
||
| "github.com/openshift/library-go/pkg/operator/events" | ||
| "github.com/openshift/library-go/pkg/operator/v1helpers" | ||
|
|
||
| "github.com/openshift/cluster-kube-apiserver-operator/pkg/operator/operatorclient" | ||
| "github.com/openshift/cluster-kube-apiserver-operator/pkg/operator/targetconfigcontroller" | ||
| ) | ||
|
|
||
| const workQueueKey = "key" | ||
|
|
||
| // CABundleController composes individual certs into CA bundle that is used | ||
| // by kube-apiserver to validate clients. | ||
| // Cert recovery refreshes "kube-control-plane-signer-ca" and needs the containing | ||
| // bundle regenerated so kube-controller-manager and kube-scheduler can connect | ||
| // using client certs. | ||
| type CABundleController struct { | ||
| configMapGetter corev1client.ConfigMapsGetter | ||
| configMapLister corev1listers.ConfigMapLister | ||
|
|
||
| eventRecorder events.Recorder | ||
|
|
||
| cachesToSync []cache.InformerSynced | ||
|
|
||
| // queue only ever has one item, but it has nice error handling backoff/retry semantics | ||
| queue workqueue.RateLimitingInterface | ||
| } | ||
|
|
||
| func NewCABundleController( | ||
| configMapGetter corev1client.ConfigMapsGetter, | ||
| kubeInformersForNamespaces v1helpers.KubeInformersForNamespaces, | ||
| eventRecorder events.Recorder, | ||
| ) (*CABundleController, error) { | ||
| c := &CABundleController{ | ||
| configMapGetter: configMapGetter, | ||
| configMapLister: kubeInformersForNamespaces.ConfigMapLister(), | ||
| eventRecorder: eventRecorder.WithComponentSuffix("manage-client-ca-bundle-recovery-controller"), | ||
| queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "CABundleRecoveryController"), | ||
| } | ||
|
|
||
| handler := cache.ResourceEventHandlerFuncs{ | ||
| AddFunc: func(obj interface{}) { c.queue.Add(workQueueKey) }, | ||
| UpdateFunc: func(old, new interface{}) { c.queue.Add(workQueueKey) }, | ||
| DeleteFunc: func(obj interface{}) { c.queue.Add(workQueueKey) }, | ||
| } | ||
|
|
||
| // we react to some config changes | ||
| namespaces := []string{ | ||
| operatorclient.GlobalUserSpecifiedConfigNamespace, | ||
| operatorclient.GlobalMachineSpecifiedConfigNamespace, | ||
| operatorclient.OperatorNamespace, | ||
| operatorclient.TargetNamespace, | ||
| } | ||
| for _, namespace := range namespaces { | ||
| informers := kubeInformersForNamespaces.InformersFor(namespace) | ||
| informers.Core().V1().ConfigMaps().Informer().AddEventHandler(handler) | ||
| c.cachesToSync = append(c.cachesToSync, informers.Core().V1().ConfigMaps().Informer().HasSynced) | ||
| } | ||
|
|
||
| return c, nil | ||
| } | ||
|
|
||
| func (c *CABundleController) Run(ctx context.Context) { | ||
| defer utilruntime.HandleCrash() | ||
|
|
||
| // FIXME: These are missing a wait group to track goroutines and handle graceful termination | ||
| // (@deads2k wants time to think it through) | ||
|
|
||
| klog.Info("Starting CA bundle controller") | ||
| defer func() { | ||
| klog.Info("Shutting down CA bundle controller") | ||
| c.queue.ShutDown() | ||
| klog.Info("CA bundle controller shut down") | ||
| }() | ||
|
|
||
| if !cache.WaitForNamedCacheSync("CABundleController", ctx.Done(), c.cachesToSync...) { | ||
| return | ||
| } | ||
|
|
||
| go func() { | ||
| wait.UntilWithContext(ctx, c.runWorker, time.Second) | ||
| }() | ||
|
|
||
| <-ctx.Done() | ||
| } | ||
|
|
||
| func (c *CABundleController) runWorker(ctx context.Context) { | ||
| for c.processNextItem(ctx) { | ||
| } | ||
| } | ||
|
|
||
| func (c *CABundleController) processNextItem(ctx context.Context) bool { | ||
| key, quit := c.queue.Get() | ||
| if quit { | ||
| return false | ||
| } | ||
| defer c.queue.Done(key) | ||
|
|
||
| err := c.sync(ctx) | ||
|
|
||
| if err == nil { | ||
| c.queue.Forget(key) | ||
| return true | ||
| } | ||
|
|
||
| utilruntime.HandleError(fmt.Errorf("%v failed with : %w", key, err)) | ||
| c.queue.AddRateLimited(key) | ||
|
|
||
| return true | ||
| } | ||
|
|
||
| func (c *CABundleController) sync(ctx context.Context) error { | ||
| // Always start 10 seconds later after a change occurred. Makes us less likely to steal work and logs from the operator. | ||
| timer := time.NewTimer(10 * time.Second) | ||
| defer timer.Stop() | ||
| select { | ||
| case <-timer.C: | ||
| case <-ctx.Done(): | ||
| return nil | ||
| } | ||
|
|
||
| _, changed, err := targetconfigcontroller.ManageClientCABundle(c.configMapLister, c.configMapGetter, c.eventRecorder) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if changed { | ||
| klog.V(2).Info("Refreshed client CA bundle.") | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| package certregenerationcontroller | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/spf13/cobra" | ||
|
|
||
| "k8s.io/client-go/kubernetes" | ||
|
|
||
| operatorv1 "github.com/openshift/api/operator/v1" | ||
| configeversionedclient "github.com/openshift/client-go/config/clientset/versioned" | ||
| configexternalinformers "github.com/openshift/client-go/config/informers/externalversions" | ||
| "github.com/openshift/library-go/pkg/controller/controllercmd" | ||
| "github.com/openshift/library-go/pkg/operator/certrotation" | ||
| "github.com/openshift/library-go/pkg/operator/genericoperatorclient" | ||
| "github.com/openshift/library-go/pkg/operator/v1helpers" | ||
|
|
||
| "github.com/openshift/cluster-kube-apiserver-operator/pkg/operator/certrotationcontroller" | ||
| "github.com/openshift/cluster-kube-apiserver-operator/pkg/operator/operatorclient" | ||
| "github.com/openshift/cluster-kube-apiserver-operator/pkg/version" | ||
| ) | ||
|
|
||
| type Options struct { | ||
| controllerContext *controllercmd.ControllerContext | ||
|
|
||
| TLSServerName string | ||
| } | ||
|
|
||
| func NewCertRegenerationControllerCommand(ctx context.Context) *cobra.Command { | ||
| o := &Options{ | ||
| TLSServerName: "localhost-recovery", | ||
| } | ||
|
|
||
| cmd := controllercmd. | ||
| NewControllerCommandConfig("cert-regeneration-controller", version.Get(), func(ctx context.Context, controllerContext *controllercmd.ControllerContext) error { | ||
| o.controllerContext = controllerContext | ||
|
|
||
| err := o.Validate(ctx) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| err = o.Complete(ctx) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| err = o.Run(ctx) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| }).NewCommandWithContext(ctx) | ||
| cmd.Use = "cert-regeneration-controller" | ||
| cmd.Short = "Start the Cluster Certificate Regeneration Controller" | ||
|
|
||
| cmd.PersistentFlags().StringVarP(&o.TLSServerName, "tls-server-name", "", o.TLSServerName, "The SNI hostname to set for the server in kubeconfig") | ||
|
|
||
| return cmd | ||
| } | ||
|
|
||
| func (o *Options) Validate(ctx context.Context) error { | ||
| return nil | ||
| } | ||
|
|
||
| func (o *Options) Complete(ctx context.Context) error { | ||
| return nil | ||
| } | ||
|
|
||
| func (o *Options) Run(ctx context.Context) error { | ||
| if len(o.TLSServerName) != 0 { | ||
| // TLSServerName chooses the SNI serving endpoint on the apiserver. | ||
| // Particularly useful when connecting to "localhost" and wanting to choose a special | ||
| // serving endpoint like "localhost-recovery" that has long-lived serving certs | ||
| // for localhost connections. | ||
| o.controllerContext.KubeConfig.TLSClientConfig.ServerName = o.TLSServerName | ||
| o.controllerContext.ProtoKubeConfig.TLSClientConfig.ServerName = o.TLSServerName | ||
| } | ||
|
|
||
| kubeClient, err := kubernetes.NewForConfig(o.controllerContext.ProtoKubeConfig) | ||
| if err != nil { | ||
| return fmt.Errorf("can't build kubernetes client: %w", err) | ||
| } | ||
|
|
||
| configClient, err := configeversionedclient.NewForConfig(o.controllerContext.KubeConfig) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create config client: %w", err) | ||
| } | ||
|
|
||
| configInformers := configexternalinformers.NewSharedInformerFactory(configClient, 10*time.Minute) | ||
|
|
||
| kubeAPIServerInformersForNamespaces := v1helpers.NewKubeInformersForNamespaces( | ||
| kubeClient, | ||
| operatorclient.GlobalMachineSpecifiedConfigNamespace, | ||
| operatorclient.GlobalUserSpecifiedConfigNamespace, | ||
| operatorclient.OperatorNamespace, | ||
| operatorclient.TargetNamespace, | ||
| ) | ||
|
|
||
| operatorClient, dynamicInformers, err := genericoperatorclient.NewStaticPodOperatorClient(o.controllerContext.KubeConfig, operatorv1.GroupVersion.WithResource("kubeapiservers")) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| certRotationScale, err := certrotation.GetCertRotationScale(kubeClient, operatorclient.GlobalUserSpecifiedConfigNamespace) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| kubeAPIServerCertRotationController, err := certrotationcontroller.NewCertRotationControllerOnlyWhenExpired( | ||
| kubeClient, | ||
| operatorClient, | ||
| configInformers, | ||
| kubeAPIServerInformersForNamespaces, | ||
| o.controllerContext.EventRecorder, | ||
| certRotationScale, | ||
| ) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| caBundleController, err := NewCABundleController( | ||
| kubeClient.CoreV1(), | ||
| kubeAPIServerInformersForNamespaces, | ||
| o.controllerContext.EventRecorder, | ||
| ) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // We can't start informers until after the resources have been requested. Now is the time. | ||
| configInformers.Start(ctx.Done()) | ||
| kubeAPIServerInformersForNamespaces.Start(ctx.Done()) | ||
| dynamicInformers.Start(ctx.Done()) | ||
|
|
||
| // FIXME: These are missing a wait group to track goroutines and handle graceful termination | ||
| // (@deads2k wants time to think it through) | ||
|
|
||
| go func() { | ||
| kubeAPIServerCertRotationController.Run(ctx, 1) | ||
| }() | ||
|
|
||
| go func() { | ||
| caBundleController.Run(ctx) | ||
| }() | ||
|
|
||
| <-ctx.Done() | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.