/* Copyright 2017 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 application import ( "context" "fmt" "net/http" "slices" "strconv" "time" "golang.org/x/time/rate" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/record" "k8s.io/client-go/util/workqueue" "k8s.io/klog/v2" v1alpha1 "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/apis/application/v1alpha1" controllers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/controllers" clientset "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/clientset/versioned" 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/application/v1alpha1" listers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/listers/application/v1alpha1" authentikapi "goauthentik.io/api/v3" ) const controllerAgentName = "application-controller" const ( SuccessSynced = "Synced" ErrResourceExists = "ErrResourceExists" MessageResourceExists = "Resource %q already exists and is not managed by Application" MessageResourceSynced = "Application synced successfully" FieldManager = controllerAgentName ) // Finalizers const ( DeleteAuthentikApplicationFinalizer = "application.t000-n.de/delete-authentik-application" ) type ApplicationController struct { kubeclientset kubernetes.Interface applicationClientset clientset.Interface authentik *authentikapi.APIClient applicationListener listers.ApplicationLister controller *controllers.Controller } func NewController( ctx context.Context, kubeclientset kubernetes.Interface, applicationClientset clientset.Interface, authentik *authentikapi.APIClient, applicationInformer informers.ApplicationInformer, ) *ApplicationController { logger := klog.FromContext(ctx) utilruntime.Must(operatorscheme.AddToScheme(scheme.Scheme)) logger.V(4).Info("Creating event broadcaster") eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx)) eventBroadcaster.StartStructuredLogging(0) eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events("")}) recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName}) ratelimiter := workqueue.NewTypedMaxOfRateLimiter( workqueue.NewTypedItemExponentialFailureRateLimiter[cache.ObjectName](5*time.Millisecond, 1000*time.Second), &workqueue.TypedBucketRateLimiter[cache.ObjectName]{Limiter: rate.NewLimiter(rate.Limit(50), 300)}, ) c := &ApplicationController{ kubeclientset: kubeclientset, applicationClientset: applicationClientset, authentik: authentik, applicationListener: applicationInformer.Lister(), } c.controller = controllers.NewController( ctx, workqueue.NewTypedRateLimitingQueue(ratelimiter), recorder, applicationInformer.Informer().HasSynced, c.syncHandler, ) logger.Info("Setting up event handlers") applicationInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: c.controller.Enqueue, UpdateFunc: func(_, newObj interface{}) { c.controller.Enqueue(newObj) }, }) return c } func (c *ApplicationController) Run(ctx context.Context, workers int) error { return c.controller.Run(ctx, workers) } func (c *ApplicationController) syncHandler(ctx context.Context, objectRef cache.ObjectName) error { logger := klog.LoggerWithValues(klog.FromContext(ctx), "objectRef", objectRef) app, err := c.applicationListener.Applications(objectRef.Namespace).Get(objectRef.Name) if err != nil { if errors.IsNotFound(err) { logger.V(4).Info("Application no longer exists") return nil } return err } logger.V(4).Info("sync Application", "name", app.Name) if !app.ObjectMeta.DeletionTimestamp.IsZero() { logger.Info("Reconciling deletion of Application", "name", app.Name) return c.reconcileDelete(ctx, app) } if app.Status.PK == "" { logger.Info("Reconciling creation of Application", "name", app.Name) return c.reconcileCreate(ctx, app) } // Check if all finalizers are present. If not, we add them. Same pattern as above, just needs a helper function to check for presence of a finalizer. if !slices.Contains(app.ObjectMeta.Finalizers, DeleteAuthentikApplicationFinalizer) { logger.Info("Ensuring finalizers are present", "name", app.Name) return c.ensureFinalizers(ctx, app) } logger.Info("Reconciling update of Application", "name", app.Name) return c.reconcileUpdate(ctx, app) } func (c *ApplicationController) ensureFinalizers(ctx context.Context, app *v1alpha1.Application) error { app.ObjectMeta.Finalizers = append(app.ObjectMeta.Finalizers, DeleteAuthentikApplicationFinalizer) return c.updateApplication(ctx, app) } func (c *ApplicationController) reconcileDelete(ctx context.Context, app *v1alpha1.Application) error { pk, err := strconv.ParseInt(app.Status.PK, 10, 32) if err != nil { return fmt.Errorf("error parsing PK: %v", err) } r, err := c.authentik.ProvidersApi.ProvidersProxyDestroy(ctx, int32(pk)).Execute() if err != nil { // This handles an edge-case, where when the ProxyProvider on Authentik has already been deleted, but the finalizer is still present. We just remove the finalizer and return. if r != nil && r.StatusCode != http.StatusNotFound { return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyDestroy`: %w with response %v", err, r) } } app.ObjectMeta.Finalizers = slices.Delete(app.ObjectMeta.Finalizers, slices.Index(app.ObjectMeta.Finalizers, DeleteAuthentikApplicationFinalizer), 1) return c.updateApplication(ctx, app) } func (c *ApplicationController) reconcileUpdate(ctx context.Context, app *v1alpha1.Application) error { _, r, err := c.authentik.CoreApi.CoreApplicationsRetrieve(ctx, app.Spec.Slug).Execute() if err != nil { if r != nil && r.StatusCode == http.StatusNotFound { // This handles an edge-case, where when the Application on Authentik has been deleted, e.g. by mistake. We just remove the PK and return. // During the next reconciliation, the Application will be re-created. app.Status.PK = "" return c.updateApplicationStatus(ctx, app) } return fmt.Errorf("error retrieving existing Application: %v with response %v", err, r) } patchedApplicationRequest := &authentikapi.PatchedApplicationRequest{ Name: &app.Spec.Name, Slug: &app.Spec.Slug, Provider: *authentikapi.NewNullableInt32(&app.Spec.Provider), } resp, r, err := c.authentik.CoreApi.CoreApplicationsPartialUpdate(ctx, app.Spec.Slug).PatchedApplicationRequest(*patchedApplicationRequest).Execute() if err != nil { return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyPartialUpdate`: %w with response %v", err, r) } app.Status.PK = resp.Pk return c.updateApplicationStatus(ctx, app) } func (c *ApplicationController) reconcileCreate(ctx context.Context, app *v1alpha1.Application) error { applicationRequest := &authentikapi.ApplicationRequest{ Name: app.Spec.Name, Slug: app.Spec.Slug, Provider: *authentikapi.NewNullableInt32(&app.Spec.Provider), } resp, r, err := c.authentik.CoreApi.CoreApplicationsCreate(ctx).ApplicationRequest(*applicationRequest).Execute() if err != nil { return fmt.Errorf("error when calling `CoreAPI.CoreApplicationsCreate`: %w with response %v", err, r) } app.Status.PK = resp.Pk return c.updateApplicationStatus(ctx, app) } func (c *ApplicationController) updateApplicationStatus(ctx context.Context, app *v1alpha1.Application) error { appCopy := app.DeepCopy() _, err := c.applicationClientset.ApplicationV1alpha1().Applications(appCopy.Namespace).UpdateStatus(ctx, appCopy, metav1.UpdateOptions{FieldManager: FieldManager}) return err } // Update metadata, spec, etc. of the Application object. func (c *ApplicationController) updateApplication(ctx context.Context, app *v1alpha1.Application) error { appCopy := app.DeepCopy() _, err := c.applicationClientset.ApplicationV1alpha1().Applications(appCopy.Namespace).Update(ctx, appCopy, metav1.UpdateOptions{FieldManager: FieldManager}) if err != nil { return fmt.Errorf("error updating Application metadata: %v", err) } return nil }