281 lines
10 KiB
Go
281 lines
10 KiB
Go
/*
|
|
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"
|
|
"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/apimachinery/pkg/util/wait"
|
|
"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"
|
|
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 Controller struct {
|
|
kubeclientset kubernetes.Interface
|
|
applicationClientset clientset.Interface
|
|
authentik *authentikapi.APIClient
|
|
|
|
applicationListener listers.ApplicationLister
|
|
applicationSynced cache.InformerSynced
|
|
|
|
workqueue workqueue.TypedRateLimitingInterface[cache.ObjectName]
|
|
recorder record.EventRecorder
|
|
}
|
|
|
|
func NewController(
|
|
ctx context.Context,
|
|
kubeclientset kubernetes.Interface,
|
|
applicationClientset clientset.Interface,
|
|
authentik *authentikapi.APIClient,
|
|
applicationInformer informers.ApplicationInformer,
|
|
) *Controller {
|
|
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 := &Controller{
|
|
kubeclientset: kubeclientset,
|
|
applicationClientset: applicationClientset,
|
|
authentik: authentik,
|
|
applicationListener: applicationInformer.Lister(),
|
|
applicationSynced: applicationInformer.Informer().HasSynced,
|
|
workqueue: workqueue.NewTypedRateLimitingQueue(ratelimiter),
|
|
recorder: recorder,
|
|
}
|
|
|
|
logger.Info("Setting up event handlers")
|
|
applicationInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
|
|
AddFunc: c.enqueueApplication,
|
|
UpdateFunc: func(_, newObj interface{}) {
|
|
c.enqueueApplication(newObj)
|
|
},
|
|
})
|
|
|
|
return c
|
|
}
|
|
|
|
func (c *Controller) Run(ctx context.Context, workers int) error {
|
|
defer utilruntime.HandleCrash()
|
|
defer c.workqueue.ShutDown()
|
|
logger := klog.FromContext(ctx)
|
|
|
|
logger.Info("Starting Application controller")
|
|
|
|
logger.Info("Waiting for informer caches to sync")
|
|
if ok := cache.WaitForCacheSync(ctx.Done(), c.applicationSynced); !ok {
|
|
return fmt.Errorf("failed to wait for caches to sync")
|
|
}
|
|
|
|
logger.Info("Starting workers", "count", workers)
|
|
for i := 0; i < workers; i++ {
|
|
go wait.UntilWithContext(ctx, c.runWorker, time.Second)
|
|
}
|
|
|
|
logger.Info("Started workers")
|
|
<-ctx.Done()
|
|
logger.Info("Shutting down workers")
|
|
return nil
|
|
}
|
|
|
|
func (c *Controller) runWorker(ctx context.Context) {
|
|
for c.processNextWorkItem(ctx) {
|
|
}
|
|
}
|
|
|
|
func (c *Controller) processNextWorkItem(ctx context.Context) bool {
|
|
objRef, shutdown := c.workqueue.Get()
|
|
logger := klog.FromContext(ctx)
|
|
if shutdown {
|
|
return false
|
|
}
|
|
defer c.workqueue.Done(objRef)
|
|
|
|
err := c.syncHandler(ctx, objRef)
|
|
if err == nil {
|
|
c.workqueue.Forget(objRef)
|
|
logger.Info("Successfully synced", "objectName", objRef)
|
|
return true
|
|
}
|
|
utilruntime.HandleErrorWithContext(ctx, err, "Error syncing; requeuing for later retry", "objectReference", objRef)
|
|
c.workqueue.AddRateLimited(objRef)
|
|
return true
|
|
}
|
|
|
|
func (c *Controller) 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 *Controller) ensureFinalizers(ctx context.Context, app *v1alpha1.Application) error {
|
|
app.ObjectMeta.Finalizers = append(app.ObjectMeta.Finalizers, DeleteAuthentikApplicationFinalizer)
|
|
return c.updateApplication(ctx, app)
|
|
}
|
|
|
|
func (c *Controller) reconcileDelete(ctx context.Context, app *v1alpha1.Application) error {
|
|
r, err := c.authentik.CoreApi.CoreApplicationsDestroy(ctx, app.Status.PK).Execute()
|
|
if err != nil {
|
|
// This handles an edge-case, where when the Application 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 `CoreAPI.CoreApplicationsDestroy`: %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 *Controller) 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 `CoreAPI.CoreApplicationsPartialUpdate`: %w with response %v", err, r)
|
|
}
|
|
|
|
app.Status.PK = resp.Pk
|
|
return c.updateApplicationStatus(ctx, app)
|
|
}
|
|
|
|
func (c *Controller) 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 *Controller) enqueueApplication(obj interface{}) {
|
|
objectRef, err := cache.ObjectToName(obj)
|
|
if err != nil {
|
|
utilruntime.HandleError(err)
|
|
return
|
|
}
|
|
c.workqueue.Add(objectRef)
|
|
}
|
|
|
|
func (c *Controller) 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 *Controller) 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
|
|
}
|