minimum runnable
This commit is contained in:
+45
-253
@@ -39,68 +39,45 @@ import (
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/klog/v2"
|
||||
|
||||
samplev1alpha1 "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/apis/samplecontroller/v1alpha1"
|
||||
clientset "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/clientset/versioned"
|
||||
samplescheme "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/clientset/versioned/scheme"
|
||||
informers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/informers/externalversions/samplecontroller/v1alpha1"
|
||||
listers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/listers/samplecontroller/v1alpha1"
|
||||
operatorscheme "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/clientset/versioned/scheme"
|
||||
informers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/informers/externalversions/proxyprovider/v1"
|
||||
listers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/listers/proxyprovider/v1"
|
||||
)
|
||||
|
||||
const controllerAgentName = "sample-controller"
|
||||
const controllerAgentName = "proxy-provider-controller"
|
||||
|
||||
const (
|
||||
// SuccessSynced is used as part of the Event 'reason' when a Foo is synced
|
||||
SuccessSynced = "Synced"
|
||||
// ErrResourceExists is used as part of the Event 'reason' when a Foo fails
|
||||
// to sync due to a Deployment of the same name already existing.
|
||||
ErrResourceExists = "ErrResourceExists"
|
||||
|
||||
// MessageResourceExists is the message used for Events when a resource
|
||||
// fails to sync due to a Deployment already existing
|
||||
MessageResourceExists = "Resource %q already exists and is not managed by Foo"
|
||||
// MessageResourceSynced is the message used for an Event fired when a Foo
|
||||
// is synced successfully
|
||||
MessageResourceSynced = "Foo synced successfully"
|
||||
// FieldManager distinguishes this controller from other things writing to API objects
|
||||
FieldManager = controllerAgentName
|
||||
SuccessSynced = "Synced"
|
||||
ErrResourceExists = "ErrResourceExists"
|
||||
MessageResourceExists = "Resource %q already exists and is not managed by ProxyProvider"
|
||||
MessageResourceSynced = "ProxyProvider synced successfully"
|
||||
FieldManager = controllerAgentName
|
||||
)
|
||||
|
||||
// Controller is the controller implementation for Foo resources
|
||||
type Controller struct {
|
||||
// kubeclientset is a standard kubernetes clientset
|
||||
kubeclientset kubernetes.Interface
|
||||
// sampleclientset is a clientset for our own API group
|
||||
sampleclientset clientset.Interface
|
||||
kubeclientset kubernetes.Interface
|
||||
operatorclientset clientset.Interface
|
||||
|
||||
deploymentsLister appslisters.DeploymentLister
|
||||
deploymentsSynced cache.InformerSynced
|
||||
foosLister listers.FooLister
|
||||
foosSynced cache.InformerSynced
|
||||
proxyLister listers.ProxyProviderLister
|
||||
proxySynced cache.InformerSynced
|
||||
|
||||
// workqueue is a rate limited work queue. This is used to queue work to be
|
||||
// processed instead of performing it as soon as a change happens. This
|
||||
// means we can ensure we only process a fixed amount of resources at a
|
||||
// time, and makes it easy to ensure we are never processing the same item
|
||||
// simultaneously in two different workers.
|
||||
workqueue workqueue.TypedRateLimitingInterface[cache.ObjectName]
|
||||
// recorder is an event recorder for recording Event resources to the
|
||||
// Kubernetes API.
|
||||
recorder record.EventRecorder
|
||||
recorder record.EventRecorder
|
||||
}
|
||||
|
||||
// NewController returns a new sample controller
|
||||
func NewController(
|
||||
ctx context.Context,
|
||||
kubeclientset kubernetes.Interface,
|
||||
sampleclientset clientset.Interface,
|
||||
operatorclientset clientset.Interface,
|
||||
deploymentInformer appsinformers.DeploymentInformer,
|
||||
fooInformer informers.FooInformer) *Controller {
|
||||
proxyInformer informers.ProxyProviderInformer,
|
||||
) *Controller {
|
||||
logger := klog.FromContext(ctx)
|
||||
|
||||
// Create event broadcaster
|
||||
// Add sample-controller types to the default Kubernetes Scheme so Events can be
|
||||
// logged for sample-controller types.
|
||||
utilruntime.Must(samplescheme.AddToScheme(scheme.Scheme))
|
||||
utilruntime.Must(operatorscheme.AddToScheme(scheme.Scheme))
|
||||
logger.V(4).Info("Creating event broadcaster")
|
||||
|
||||
eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx))
|
||||
@@ -112,70 +89,53 @@ func NewController(
|
||||
&workqueue.TypedBucketRateLimiter[cache.ObjectName]{Limiter: rate.NewLimiter(rate.Limit(50), 300)},
|
||||
)
|
||||
|
||||
controller := &Controller{
|
||||
c := &Controller{
|
||||
kubeclientset: kubeclientset,
|
||||
sampleclientset: sampleclientset,
|
||||
operatorclientset: operatorclientset,
|
||||
deploymentsLister: deploymentInformer.Lister(),
|
||||
deploymentsSynced: deploymentInformer.Informer().HasSynced,
|
||||
foosLister: fooInformer.Lister(),
|
||||
foosSynced: fooInformer.Informer().HasSynced,
|
||||
proxyLister: proxyInformer.Lister(),
|
||||
proxySynced: proxyInformer.Informer().HasSynced,
|
||||
workqueue: workqueue.NewTypedRateLimitingQueue(ratelimiter),
|
||||
recorder: recorder,
|
||||
}
|
||||
|
||||
logger.Info("Setting up event handlers")
|
||||
// Set up an event handler for when Foo resources change
|
||||
fooInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: controller.enqueueFoo,
|
||||
UpdateFunc: func(old, new interface{}) {
|
||||
controller.enqueueFoo(new)
|
||||
proxyInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: c.enqueueProxyProvider,
|
||||
UpdateFunc: func(_, newObj interface{}) {
|
||||
c.enqueueProxyProvider(newObj)
|
||||
},
|
||||
})
|
||||
// Set up an event handler for when Deployment resources change. This
|
||||
// handler will lookup the owner of the given Deployment, and if it is
|
||||
// owned by a Foo resource then the handler will enqueue that Foo resource for
|
||||
// processing. This way, we don't need to implement custom logic for
|
||||
// handling Deployment resources. More info on this pattern:
|
||||
// https://github.com/kubernetes/community/blob/8cafef897a22026d42f5e5bb3f104febe7e29830/contributors/devel/controllers.md
|
||||
deploymentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: controller.handleObject,
|
||||
AddFunc: c.handleObject,
|
||||
UpdateFunc: func(old, new interface{}) {
|
||||
newDepl := new.(*appsv1.Deployment)
|
||||
oldDepl := old.(*appsv1.Deployment)
|
||||
if newDepl.ResourceVersion == oldDepl.ResourceVersion {
|
||||
// Periodic resync will send update events for all known Deployments.
|
||||
// Two different versions of the same Deployment will always have different RVs.
|
||||
return
|
||||
}
|
||||
controller.handleObject(new)
|
||||
c.handleObject(new)
|
||||
},
|
||||
DeleteFunc: controller.handleObject,
|
||||
DeleteFunc: c.handleObject,
|
||||
})
|
||||
|
||||
return controller
|
||||
return c
|
||||
}
|
||||
|
||||
// Run will set up the event handlers for types we are interested in, as well
|
||||
// as syncing informer caches and starting workers. It will block until stopCh
|
||||
// is closed, at which point it will shutdown the workqueue and wait for
|
||||
// workers to finish processing their current work items.
|
||||
func (c *Controller) Run(ctx context.Context, workers int) error {
|
||||
defer utilruntime.HandleCrash()
|
||||
defer c.workqueue.ShutDown()
|
||||
logger := klog.FromContext(ctx)
|
||||
|
||||
// Start the informer factories to begin populating the informer caches
|
||||
logger.Info("Starting Foo controller")
|
||||
logger.Info("Starting ProxyProvider controller")
|
||||
|
||||
// Wait for the caches to be synced before starting workers
|
||||
logger.Info("Waiting for informer caches to sync")
|
||||
|
||||
if ok := cache.WaitForCacheSync(ctx.Done(), c.deploymentsSynced, c.foosSynced); !ok {
|
||||
if ok := cache.WaitForCacheSync(ctx.Done(), c.deploymentsSynced, c.proxySynced); !ok {
|
||||
return fmt.Errorf("failed to wait for caches to sync")
|
||||
}
|
||||
|
||||
logger.Info("Starting workers", "count", workers)
|
||||
// Launch two workers to process Foo resources
|
||||
for i := 0; i < workers; i++ {
|
||||
go wait.UntilWithContext(ctx, c.runWorker, time.Second)
|
||||
}
|
||||
@@ -183,239 +143,71 @@ func (c *Controller) Run(ctx context.Context, workers int) error {
|
||||
logger.Info("Started workers")
|
||||
<-ctx.Done()
|
||||
logger.Info("Shutting down workers")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// runWorker is a long-running function that will continually call the
|
||||
// processNextWorkItem function in order to read and process a message on the
|
||||
// workqueue.
|
||||
func (c *Controller) runWorker(ctx context.Context) {
|
||||
for c.processNextWorkItem(ctx) {
|
||||
}
|
||||
}
|
||||
|
||||
// processNextWorkItem will read a single work item off the workqueue and
|
||||
// attempt to process it, by calling the syncHandler.
|
||||
func (c *Controller) processNextWorkItem(ctx context.Context) bool {
|
||||
objRef, shutdown := c.workqueue.Get()
|
||||
logger := klog.FromContext(ctx)
|
||||
|
||||
if shutdown {
|
||||
return false
|
||||
}
|
||||
|
||||
// We call Done at the end of this func so the workqueue knows we have
|
||||
// finished processing this item. We also must remember to call Forget
|
||||
// if we do not want this work item being re-queued. For example, we do
|
||||
// not call Forget if a transient error occurs, instead the item is
|
||||
// put back on the workqueue and attempted again after a back-off
|
||||
// period.
|
||||
defer c.workqueue.Done(objRef)
|
||||
|
||||
// Run the syncHandler, passing it the structured reference to the object to be synced.
|
||||
err := c.syncHandler(ctx, objRef)
|
||||
if err == nil {
|
||||
// If no error occurs then we Forget this item so it does not
|
||||
// get queued again until another change happens.
|
||||
c.workqueue.Forget(objRef)
|
||||
logger.Info("Successfully synced", "objectName", objRef)
|
||||
return true
|
||||
}
|
||||
// there was a failure so be sure to report it. This method allows for
|
||||
// pluggable error handling which can be used for things like
|
||||
// cluster-monitoring.
|
||||
utilruntime.HandleErrorWithContext(ctx, err, "Error syncing; requeuing for later retry", "objectReference", objRef)
|
||||
// since we failed, we should requeue the item to work on later. This
|
||||
// method will add a backoff to avoid hotlooping on particular items
|
||||
// (they're probably still not going to work right away) and overall
|
||||
// controller protection (everything I've done is broken, this controller
|
||||
// needs to calm down or it can starve other useful work) cases.
|
||||
c.workqueue.AddRateLimited(objRef)
|
||||
return true
|
||||
}
|
||||
|
||||
// syncHandler compares the actual state with the desired, and attempts to
|
||||
// converge the two. It then updates the Status block of the Foo resource
|
||||
// with the current status of the resource.
|
||||
func (c *Controller) syncHandler(ctx context.Context, objectRef cache.ObjectName) error {
|
||||
logger := klog.LoggerWithValues(klog.FromContext(ctx), "objectRef", objectRef)
|
||||
|
||||
// Get the Foo resource with this namespace/name
|
||||
foo, err := c.foosLister.Foos(objectRef.Namespace).Get(objectRef.Name)
|
||||
pp, err := c.proxyLister.ProxyProviders(objectRef.Namespace).Get(objectRef.Name)
|
||||
if err != nil {
|
||||
// The Foo resource may no longer exist, in which case we stop
|
||||
// processing.
|
||||
if errors.IsNotFound(err) {
|
||||
utilruntime.HandleErrorWithContext(ctx, err, "Foo referenced by item in work queue no longer exists", "objectReference", objectRef)
|
||||
logger.V(4).Info("ProxyProvider no longer exists")
|
||||
return nil
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
deploymentName := foo.Spec.DeploymentName
|
||||
if deploymentName == "" {
|
||||
// We choose to absorb the error here as the worker would requeue the
|
||||
// resource otherwise. Instead, the next time the resource is updated
|
||||
// the resource will be queued again.
|
||||
utilruntime.HandleErrorWithContext(ctx, nil, "Deployment name missing from object reference", "objectReference", objectRef)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Get the deployment with the name specified in Foo.spec
|
||||
deployment, err := c.deploymentsLister.Deployments(foo.Namespace).Get(deploymentName)
|
||||
// If the resource doesn't exist, we'll create it
|
||||
if errors.IsNotFound(err) {
|
||||
deployment, err = c.kubeclientset.AppsV1().Deployments(foo.Namespace).Create(ctx, newDeployment(foo), metav1.CreateOptions{FieldManager: FieldManager})
|
||||
}
|
||||
|
||||
// If an error occurs during Get/Create, we'll requeue the item so we can
|
||||
// attempt processing again later. This could have been caused by a
|
||||
// temporary network failure, or any other transient reason.
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// If the Deployment is not controlled by this Foo resource, we should log
|
||||
// a warning to the event recorder and return error msg.
|
||||
if !metav1.IsControlledBy(deployment, foo) {
|
||||
msg := fmt.Sprintf(MessageResourceExists, deployment.Name)
|
||||
c.recorder.Event(foo, corev1.EventTypeWarning, ErrResourceExists, msg)
|
||||
return fmt.Errorf("%s", msg)
|
||||
}
|
||||
|
||||
// If this number of the replicas on the Foo resource is specified, and the
|
||||
// number does not equal the current desired replicas on the Deployment, we
|
||||
// should update the Deployment resource.
|
||||
if foo.Spec.Replicas != nil && *foo.Spec.Replicas != *deployment.Spec.Replicas {
|
||||
logger.V(4).Info("Update deployment resource", "currentReplicas", *deployment.Spec.Replicas, "desiredReplicas", *foo.Spec.Replicas)
|
||||
deployment, err = c.kubeclientset.AppsV1().Deployments(foo.Namespace).Update(ctx, newDeployment(foo), metav1.UpdateOptions{FieldManager: FieldManager})
|
||||
}
|
||||
|
||||
// If an error occurs during Update, we'll requeue the item so we can
|
||||
// attempt processing again later. This could have been caused by a
|
||||
// temporary network failure, or any other transient reason.
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Finally, we update the status block of the Foo resource to reflect the
|
||||
// current state of the world
|
||||
err = c.updateFooStatus(ctx, foo, deployment)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.recorder.Event(foo, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
|
||||
logger.V(4).Info("sync ProxyProvider", "name", pp.Name)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Controller) updateFooStatus(ctx context.Context, foo *samplev1alpha1.Foo, deployment *appsv1.Deployment) error {
|
||||
// NEVER modify objects from the store. It's a read-only, local cache.
|
||||
// You can use DeepCopy() to make a deep copy of original object and modify this copy
|
||||
// Or create a copy manually for better performance
|
||||
fooCopy := foo.DeepCopy()
|
||||
fooCopy.Status.AvailableReplicas = deployment.Status.AvailableReplicas
|
||||
// If the CustomResourceSubresources feature gate is not enabled,
|
||||
// we must use Update instead of UpdateStatus to update the Status block of the Foo resource.
|
||||
// UpdateStatus will not allow changes to the Spec of the resource,
|
||||
// which is ideal for ensuring nothing other than resource status has been updated.
|
||||
_, err := c.sampleclientset.SamplecontrollerV1alpha1().Foos(foo.Namespace).UpdateStatus(ctx, fooCopy, metav1.UpdateOptions{FieldManager: FieldManager})
|
||||
return err
|
||||
}
|
||||
|
||||
// enqueueFoo takes a Foo resource and converts it into a namespace/name
|
||||
// string which is then put onto the work queue. This method should *not* be
|
||||
// passed resources of any type other than Foo.
|
||||
func (c *Controller) enqueueFoo(obj interface{}) {
|
||||
if objectRef, err := cache.ObjectToName(obj); err != nil {
|
||||
func (c *Controller) enqueueProxyProvider(obj interface{}) {
|
||||
objectRef, err := cache.ObjectToName(obj)
|
||||
if err != nil {
|
||||
utilruntime.HandleError(err)
|
||||
return
|
||||
} else {
|
||||
c.workqueue.Add(objectRef)
|
||||
}
|
||||
c.workqueue.Add(objectRef)
|
||||
}
|
||||
|
||||
// handleObject will take any resource implementing metav1.Object and attempt
|
||||
// to find the Foo resource that 'owns' it. It does this by looking at the
|
||||
// objects metadata.ownerReferences field for an appropriate OwnerReference.
|
||||
// It then enqueues that Foo resource to be processed. If the object does not
|
||||
// have an appropriate OwnerReference, it will simply be skipped.
|
||||
func (c *Controller) handleObject(obj interface{}) {
|
||||
var object metav1.Object
|
||||
var ok bool
|
||||
logger := klog.FromContext(context.Background())
|
||||
if object, ok = obj.(metav1.Object); !ok {
|
||||
// Optional: resolve Deployment owners back to ProxyProvider and enqueue.
|
||||
_, ok := obj.(metav1.Object)
|
||||
if !ok {
|
||||
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
|
||||
if !ok {
|
||||
// If the object value is not too big and does not contain sensitive information then
|
||||
// it may be useful to include it.
|
||||
utilruntime.HandleErrorWithContext(context.Background(), nil, "Error decoding object, invalid type", "type", fmt.Sprintf("%T", obj))
|
||||
utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
|
||||
return
|
||||
}
|
||||
object, ok = tombstone.Obj.(metav1.Object)
|
||||
_, ok = tombstone.Obj.(metav1.Object)
|
||||
if !ok {
|
||||
// If the object value is not too big and does not contain sensitive information then
|
||||
// it may be useful to include it.
|
||||
utilruntime.HandleErrorWithContext(context.Background(), nil, "Error decoding object tombstone, invalid type", "type", fmt.Sprintf("%T", tombstone.Obj))
|
||||
utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a metav1.Object %#v", obj))
|
||||
return
|
||||
}
|
||||
logger.V(4).Info("Recovered deleted object", "resourceName", object.GetName())
|
||||
}
|
||||
logger.V(4).Info("Processing object", "object", klog.KObj(object))
|
||||
if ownerRef := metav1.GetControllerOf(object); ownerRef != nil {
|
||||
// If this object is not owned by a Foo, we should not do anything more
|
||||
// with it.
|
||||
if ownerRef.Kind != "Foo" {
|
||||
return
|
||||
}
|
||||
|
||||
foo, err := c.foosLister.Foos(object.GetNamespace()).Get(ownerRef.Name)
|
||||
if err != nil {
|
||||
logger.V(4).Info("Ignore orphaned object", "object", klog.KObj(object), "foo", ownerRef.Name)
|
||||
return
|
||||
}
|
||||
|
||||
c.enqueueFoo(foo)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// newDeployment creates a new Deployment for a Foo resource. It also sets
|
||||
// the appropriate OwnerReferences on the resource so handleObject can discover
|
||||
// the Foo resource that 'owns' it.
|
||||
func newDeployment(foo *samplev1alpha1.Foo) *appsv1.Deployment {
|
||||
labels := map[string]string{
|
||||
"app": "nginx",
|
||||
"controller": foo.Name,
|
||||
}
|
||||
return &appsv1.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: foo.Spec.DeploymentName,
|
||||
Namespace: foo.Namespace,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(foo, samplev1alpha1.SchemeGroupVersion.WithKind("Foo")),
|
||||
},
|
||||
},
|
||||
Spec: appsv1.DeploymentSpec{
|
||||
Replicas: foo.Spec.Replicas,
|
||||
Selector: &metav1.LabelSelector{
|
||||
MatchLabels: labels,
|
||||
},
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Labels: labels,
|
||||
},
|
||||
Spec: corev1.PodSpec{
|
||||
Containers: []corev1.Container{
|
||||
{
|
||||
Name: "nginx",
|
||||
Image: "nginx:latest",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user