feat: add bare application controller

This commit is contained in:
2026-05-17 19:17:22 +02:00
parent a85339cf3e
commit 0d84611b6c
31 changed files with 2161 additions and 19 deletions
+280
View File
@@ -0,0 +1,280 @@
/*
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.Spec.Slug).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
}
@@ -0,0 +1,349 @@
// AI generated tests and not yet reviewed.
package application
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"net/url"
"slices"
"strings"
"testing"
v1alpha1 "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/apis/application/v1alpha1"
operatorfake "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/clientset/versioned/fake"
operatorinformers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/informers/externalversions"
authentikapi "goauthentik.io/api/v3"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/tools/cache"
)
func TestController_syncHandler_create(t *testing.T) {
const wantPK = "42"
server := newAuthentikTestServer(t, authentikTestHandlers{
applicationCreate: func(w http.ResponseWriter, _ *http.Request) {
writeJSON(t, w, http.StatusCreated, map[string]any{"pk": wantPK})
},
})
t.Cleanup(server.Close)
ctrl, ctx, cancel := newTestController(t, testApplication(), server.URL)
t.Cleanup(cancel)
err := ctrl.syncHandler(ctx, cache.ObjectName{Namespace: "default", Name: "test-app"})
if err != nil {
t.Fatalf("syncHandler() error = %v", err)
}
got := getApplication(t, ctrl, "default", "test-app")
if got.Status.PK != wantPK {
t.Fatalf("status.pk = %q, want %q", got.Status.PK, wantPK)
}
}
func TestController_syncHandler_ensureFinalizers(t *testing.T) {
app := testApplication()
app.Status.PK = "42"
server := newAuthentikTestServer(t, authentikTestHandlers{})
t.Cleanup(server.Close)
ctrl, ctx, cancel := newTestController(t, app, server.URL)
t.Cleanup(cancel)
err := ctrl.syncHandler(ctx, cache.ObjectName{Namespace: app.Namespace, Name: app.Name})
if err != nil {
t.Fatalf("syncHandler() error = %v", err)
}
got := getApplication(t, ctrl, app.Namespace, app.Name)
if !slices.Contains(got.Finalizers, DeleteAuthentikApplicationFinalizer) {
t.Fatalf("finalizers = %v, want %q", got.Finalizers, DeleteAuthentikApplicationFinalizer)
}
}
func TestController_syncHandler_update(t *testing.T) {
app := testApplication()
app.Status.PK = "42"
app.Finalizers = []string{DeleteAuthentikApplicationFinalizer}
server := newAuthentikTestServer(t, authentikTestHandlers{
applicationRetrieve: func(w http.ResponseWriter, _ *http.Request) {
writeJSON(t, w, http.StatusOK, map[string]any{"pk": "42"})
},
applicationPartialUpdate: func(w http.ResponseWriter, _ *http.Request) {
writeJSON(t, w, http.StatusOK, map[string]any{"pk": "42"})
},
})
t.Cleanup(server.Close)
ctrl, ctx, cancel := newTestController(t, app, server.URL)
t.Cleanup(cancel)
err := ctrl.syncHandler(ctx, cache.ObjectName{Namespace: app.Namespace, Name: app.Name})
if err != nil {
t.Fatalf("syncHandler() error = %v", err)
}
got := getApplication(t, ctrl, app.Namespace, app.Name)
if got.Status.PK != "42" {
t.Fatalf("status.pk = %q, want 42", got.Status.PK)
}
}
func TestController_syncHandler_update_applicationNotFound(t *testing.T) {
app := testApplication()
app.Status.PK = "42"
app.Finalizers = []string{DeleteAuthentikApplicationFinalizer}
server := newAuthentikTestServer(t, authentikTestHandlers{
applicationRetrieve: func(w http.ResponseWriter, _ *http.Request) {
http.NotFound(w, nil)
},
})
t.Cleanup(server.Close)
ctrl, ctx, cancel := newTestController(t, app, server.URL)
t.Cleanup(cancel)
err := ctrl.syncHandler(ctx, cache.ObjectName{Namespace: app.Namespace, Name: app.Name})
if err != nil {
t.Fatalf("syncHandler() error = %v", err)
}
got := getApplication(t, ctrl, app.Namespace, app.Name)
if got.Status.PK != "" {
t.Fatalf("status.pk = %q, want empty after application not found", got.Status.PK)
}
}
func TestController_syncHandler_delete(t *testing.T) {
now := metav1.Now()
app := testApplication()
app.Status.PK = "42"
app.DeletionTimestamp = &now
app.Finalizers = []string{DeleteAuthentikApplicationFinalizer}
var destroyCalled bool
server := newAuthentikTestServer(t, authentikTestHandlers{
applicationDestroy: func(w http.ResponseWriter, r *http.Request) {
destroyCalled = true
if r.Method != http.MethodDelete {
t.Errorf("destroy method = %s, want DELETE", r.Method)
}
w.WriteHeader(http.StatusNoContent)
},
})
t.Cleanup(server.Close)
ctrl, ctx, cancel := newTestController(t, app, server.URL)
t.Cleanup(cancel)
err := ctrl.syncHandler(ctx, cache.ObjectName{Namespace: app.Namespace, Name: app.Name})
if err != nil {
t.Fatalf("syncHandler() error = %v", err)
}
if !destroyCalled {
t.Fatal("expected Authentik destroy call")
}
got := getApplication(t, ctrl, app.Namespace, app.Name)
if slices.Contains(got.Finalizers, DeleteAuthentikApplicationFinalizer) {
t.Fatalf("finalizers = %v, want finalizer removed", got.Finalizers)
}
}
func TestController_syncHandler_delete_providerAlreadyGone(t *testing.T) {
now := metav1.Now()
app := testApplication()
app.Status.PK = "42"
app.DeletionTimestamp = &now
app.Finalizers = []string{DeleteAuthentikApplicationFinalizer}
server := newAuthentikTestServer(t, authentikTestHandlers{
applicationDestroy: func(w http.ResponseWriter, _ *http.Request) {
http.NotFound(w, nil)
},
})
t.Cleanup(server.Close)
ctrl, ctx, cancel := newTestController(t, app, server.URL)
t.Cleanup(cancel)
err := ctrl.syncHandler(ctx, cache.ObjectName{Namespace: app.Namespace, Name: app.Name})
if err != nil {
t.Fatalf("syncHandler() error = %v", err)
}
got := getApplication(t, ctrl, app.Namespace, app.Name)
if slices.Contains(got.Finalizers, DeleteAuthentikApplicationFinalizer) {
t.Fatalf("finalizers = %v, want finalizer removed after 404", got.Finalizers)
}
}
func TestController_syncHandler_notFound(t *testing.T) {
server := newAuthentikTestServer(t, authentikTestHandlers{})
t.Cleanup(server.Close)
ctrl, ctx, cancel := newTestController(t, nil, server.URL)
t.Cleanup(cancel)
err := ctrl.syncHandler(ctx, cache.ObjectName{Namespace: "default", Name: "missing"})
if err != nil {
t.Fatalf("syncHandler() error = %v, want nil for missing object", err)
}
}
// --- test helpers ---
func testApplication() *v1alpha1.Application {
return &v1alpha1.Application{
TypeMeta: metav1.TypeMeta{
APIVersion: v1alpha1.SchemeGroupVersion.String(),
Kind: "Application",
},
ObjectMeta: metav1.ObjectMeta{
Name: "test-app",
Namespace: "default",
},
Spec: v1alpha1.ApplicationSpec{
Name: "My Application",
Slug: "my-app",
Provider: 7,
},
}
}
func newTestController(t *testing.T, app *v1alpha1.Application, authentikURL string) (*Controller, context.Context, context.CancelFunc) {
t.Helper()
ctx, cancel := context.WithCancel(context.Background())
ctrl, _, stop := newTestControllerWithContext(t, ctx, app, authentikURL)
return ctrl, ctx, func() {
cancel()
stop()
}
}
func newTestControllerWithContext(t *testing.T, ctx context.Context, app *v1alpha1.Application, authentikURL string) (*Controller, context.Context, func()) {
t.Helper()
authentikClient := newAuthentikAPIClientForTest(t, authentikURL)
var objects []runtime.Object
if app != nil {
objects = append(objects, app)
}
applicationClient := operatorfake.NewSimpleClientset(objects...)
informerFactory := operatorinformers.NewSharedInformerFactory(applicationClient, 0)
applicationInformer := informerFactory.Application().V1alpha1().Applications()
ctrl := NewController(ctx, fake.NewClientset(), applicationClient, authentikClient, applicationInformer)
informerFactory.Start(ctx.Done())
for informerType, synced := range informerFactory.WaitForCacheSync(ctx.Done()) {
if !synced {
t.Fatalf("informer %v failed to sync", informerType)
}
}
return ctrl, ctx, func() {}
}
func newAuthentikAPIClientForTest(t *testing.T, serverURL string) *authentikapi.APIClient {
t.Helper()
u, err := url.Parse(serverURL)
if err != nil {
t.Fatalf("parse server URL: %v", err)
}
cfg := authentikapi.NewConfiguration()
cfg.Scheme = u.Scheme
cfg.Host = u.Host
return authentikapi.NewAPIClient(cfg)
}
type authentikTestHandlers struct {
applicationCreate http.HandlerFunc
applicationRetrieve http.HandlerFunc
applicationPartialUpdate http.HandlerFunc
applicationDestroy http.HandlerFunc
}
func newAuthentikTestServer(t *testing.T, handlers authentikTestHandlers) *httptest.Server {
t.Helper()
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
path := r.URL.Path
switch {
case path == "/api/v3/core/applications/" && r.Method == http.MethodPost:
if handlers.applicationCreate != nil {
handlers.applicationCreate(w, r)
return
}
http.NotFound(w, r)
case strings.HasPrefix(path, "/api/v3/core/applications/") && strings.HasSuffix(path, "/"):
slugPath := strings.TrimPrefix(path, "/api/v3/core/applications/")
if slugPath == "" {
http.NotFound(w, r)
return
}
switch r.Method {
case http.MethodGet:
if handlers.applicationRetrieve != nil {
handlers.applicationRetrieve(w, r)
return
}
http.NotFound(w, r)
case http.MethodPatch:
if handlers.applicationPartialUpdate != nil {
handlers.applicationPartialUpdate(w, r)
return
}
http.NotFound(w, r)
case http.MethodDelete:
if handlers.applicationDestroy != nil {
handlers.applicationDestroy(w, r)
return
}
http.NotFound(w, r)
default:
http.Error(w, "unexpected method on application instance", http.StatusMethodNotAllowed)
}
default:
http.NotFound(w, r)
}
})
return httptest.NewServer(handler)
}
func writeJSON(t *testing.T, w http.ResponseWriter, status int, body any) {
t.Helper()
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
if err := json.NewEncoder(w).Encode(body); err != nil {
t.Fatalf("write JSON response: %v", err)
}
}
func getApplication(t *testing.T, ctrl *Controller, namespace, name string) *v1alpha1.Application {
t.Helper()
got, err := ctrl.applicationClientset.ApplicationV1alpha1().Applications(namespace).Get(
context.Background(), name, metav1.GetOptions{},
)
if err != nil {
t.Fatalf("get Application: %v", err)
}
return got
}