refactor: switch to reconsiliation pattern

artifacts/crd/crd.yaml

@@ -2,6 +2,8 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   name: proxyproviders.proxyprovider.t000-n.de
+  finalizers:
+    - proxyprovider.t000-n.de/delete-authentik-proxyprovider
 spec:
   group: proxyprovider.t000-n.de
   versions:

controller.go

@@ -19,22 +19,21 @@ package main
 import (
 	"context"
 	"fmt"
+	"net/http"
+	"slices"
 	"strconv"
 	"time"
 
 	"golang.org/x/time/rate"
 
-	appsv1 "k8s.io/api/apps/v1"
 	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"
-	appsinformers "k8s.io/client-go/informers/apps/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
-	appslisters "k8s.io/client-go/listers/apps/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
@@ -58,15 +57,18 @@ const (
 	FieldManager          = controllerAgentName
 )
 
+// Finalizers
+const (
+	DeleteAuthentikProxyProviderFinalizer = "proxyprovider.t000-n.de/delete-authentik-proxyprovider"
+)
+
 type Controller struct {
 	kubeclientset          kubernetes.Interface
 	proxyProviderClientset clientset.Interface
 	authentik              *authentikapi.APIClient
 
-	deploymentsLister appslisters.DeploymentLister
+	proxyLister listers.ProxyProviderLister
-	deploymentsSynced cache.InformerSynced
+	proxySynced cache.InformerSynced
-	proxyLister       listers.ProxyProviderLister
-	proxySynced       cache.InformerSynced
 
 	workqueue workqueue.TypedRateLimitingInterface[cache.ObjectName]
 	recorder  record.EventRecorder
@@ -77,7 +79,6 @@ func NewController(
 	kubeclientset kubernetes.Interface,
 	proxyProviderClientset clientset.Interface,
 	authentik *authentikapi.APIClient,
-	deploymentInformer appsinformers.DeploymentInformer,
 	proxyInformer informers.ProxyProviderInformer,
 ) *Controller {
 	logger := klog.FromContext(ctx)
@@ -98,8 +99,6 @@ func NewController(
 		kubeclientset:          kubeclientset,
 		proxyProviderClientset: proxyProviderClientset,
 		authentik:              authentik,
-		deploymentsLister:      deploymentInformer.Lister(),
-		deploymentsSynced:      deploymentInformer.Informer().HasSynced,
 		proxyLister:            proxyInformer.Lister(),
 		proxySynced:            proxyInformer.Informer().HasSynced,
 		workqueue:              workqueue.NewTypedRateLimitingQueue(ratelimiter),
@@ -113,18 +112,6 @@ func NewController(
 			c.enqueueProxyProvider(newObj)
 		},
 	})
-	deploymentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: c.handleObject,
-		UpdateFunc: func(old, new interface{}) {
-			newDepl := new.(*appsv1.Deployment)
-			oldDepl := old.(*appsv1.Deployment)
-			if newDepl.ResourceVersion == oldDepl.ResourceVersion {
-				return
-			}
-			c.handleObject(new)
-		},
-		DeleteFunc: c.handleObject,
-	})
 
 	return c
 }
@@ -137,7 +124,7 @@ func (c *Controller) Run(ctx context.Context, workers int) error {
 	logger.Info("Starting ProxyProvider controller")
 
 	logger.Info("Waiting for informer caches to sync")
-	if ok := cache.WaitForCacheSync(ctx.Done(), c.deploymentsSynced, c.proxySynced); !ok {
+	if ok := cache.WaitForCacheSync(ctx.Done(), c.proxySynced); !ok {
 		return fmt.Errorf("failed to wait for caches to sync")
 	}
 
@@ -189,54 +176,97 @@ func (c *Controller) syncHandler(ctx context.Context, objectRef cache.ObjectName
 	}
 	logger.V(4).Info("sync ProxyProvider", "name", pp.Name)
 
-	if pp.Status.PK != "" {
+	if !pp.ObjectMeta.DeletionTimestamp.IsZero() {
-		// We retrieve the existing PP from the API by slug.
+		logger.Info("Reconciling deletion of ProxyProvider", "name", pp.Name)
-		pk, err := strconv.ParseInt(pp.Status.PK, 10, 32)
+		return c.reconcileDelete(ctx, pp)
-		if err != nil {
+	}
-			return fmt.Errorf("error parsing PK: %v", err)
-		}
-		_, _, err = c.authentik.ProvidersApi.ProvidersAllRetrieve(ctx, int32(pk)).Execute()
-		if err != nil {
-			return fmt.Errorf("error retrieving existing ProxyProvider: %v", err)
-		}
 
-		// We update the existing PP with the new spec.
+	if pp.Status.PK == "" {
-		proxyProviderRequest := &authentikapi.ProxyProviderRequest{
+		logger.Info("Reconciling creation of ProxyProvider", "name", pp.Name)
-			Name:              pp.Spec.Name,
+		return c.reconcileCreate(ctx, pp)
-			AuthorizationFlow: pp.Spec.AuthorizationFlow,
+	}
-			InvalidationFlow:  pp.Spec.InvalidationFlow,
+
-			ExternalHost:      pp.Spec.ExternalHost,
+	// 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.
-			Mode:              authentikapi.PROXYMODE_FORWARD_SINGLE.Ptr(),
+	if !slices.Contains(pp.ObjectMeta.Finalizers, DeleteAuthentikProxyProviderFinalizer) {
-		}
+		logger.Info("Ensuring finalizers are present", "name", pp.Name)
-		resp, r, err := c.authentik.ProvidersApi.ProvidersProxyUpdate(ctx, int32(pk)).ProxyProviderRequest(*proxyProviderRequest).Execute()
+		return c.ensureFinalizers(ctx, pp)
-		if err != nil {
+	}
-			return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyUpdate`: %w with response %v", err, r)
+
-		}
+	logger.Info("Reconciling update of ProxyProvider", "name", pp.Name)
-		pp.Status.PK = strconv.Itoa(int(resp.Pk))
+	return c.reconcileUpdate(ctx, pp)
-		err = c.updateProxyProviderStatus(ctx, pp)
+}
-		if err != nil {
+
-			return fmt.Errorf("error updating ProxyProvider status: %v", err)
+func (c *Controller) ensureFinalizers(ctx context.Context, pp *v1.ProxyProvider) error {
-		}
+	pp.ObjectMeta.Finalizers = append(pp.ObjectMeta.Finalizers, DeleteAuthentikProxyProviderFinalizer)
-	} else {
+	return c.updateProxyProvider(ctx, pp)
-		proxyProviderRequest := &authentikapi.ProxyProviderRequest{
+}
-			Name:              pp.Spec.Name,
+
-			AuthorizationFlow: pp.Spec.AuthorizationFlow,
+func (c *Controller) reconcileDelete(ctx context.Context, pp *v1.ProxyProvider) error {
-			InvalidationFlow:  pp.Spec.InvalidationFlow,
+	pk, err := strconv.ParseInt(pp.Status.PK, 10, 32)
-			ExternalHost:      pp.Spec.ExternalHost,
+	if err != nil {
-			Mode:              authentikapi.PROXYMODE_FORWARD_SINGLE.Ptr(),
+		return fmt.Errorf("error parsing PK: %v", err)
-		}
+	}
-		resp, r, err := c.authentik.ProvidersApi.ProvidersProxyCreate(ctx).ProxyProviderRequest(*proxyProviderRequest).Execute()
+
-		if err != nil {
+	r, err := c.authentik.ProvidersApi.ProvidersProxyDestroy(ctx, int32(pk)).Execute()
-			return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyCreate`: %w with response %v", err, r)
+	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.
-		pp.Status.PK = strconv.Itoa(int(resp.Pk))
+		if r.StatusCode != http.StatusNotFound {
-		err = c.updateProxyProviderStatus(ctx, pp)
+			return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyDestroy`: %w with response %v", err, r)
-		if err != nil {
-			return fmt.Errorf("error updating ProxyProvider status: %v", err)
 		}
 	}
 
-	return nil
+	pp.ObjectMeta.Finalizers = slices.Delete(pp.ObjectMeta.Finalizers, slices.Index(pp.ObjectMeta.Finalizers, DeleteAuthentikProxyProviderFinalizer), 1)
+	return c.updateProxyProvider(ctx, pp)
+}
+
+func (c *Controller) reconcileUpdate(ctx context.Context, pp *v1.ProxyProvider) error {
+	// We retrieve the existing PP from the API by slug.
+	pk, err := strconv.ParseInt(pp.Status.PK, 10, 32)
+	if err != nil {
+		return fmt.Errorf("error parsing PK: %v", err)
+	}
+	_, r, err := c.authentik.ProvidersApi.ProvidersAllRetrieve(ctx, int32(pk)).Execute()
+	if err != nil && r.StatusCode != http.StatusNotFound {
+
+		return fmt.Errorf("error retrieving existing ProxyProvider: %v with response %v", err, r)
+	} else if r.StatusCode == http.StatusNotFound {
+		// This handles an edge-case, where when the PorxyProvider on Authentik has been deleted, e.g. by mistake. We just remove the PK and return.
+		// During the next reconciliation, the ProxyProvider will be re-created.
+		pp.Status.PK = ""
+	}
+
+	proxyProviderRequest := &authentikapi.PatchedProxyProviderRequest{
+		Name:              &pp.Spec.Name,
+		AuthorizationFlow: &pp.Spec.AuthorizationFlow,
+		InvalidationFlow:  &pp.Spec.InvalidationFlow,
+		ExternalHost:      &pp.Spec.ExternalHost,
+		Mode:              authentikapi.PROXYMODE_FORWARD_SINGLE.Ptr(),
+	}
+	resp, r, err := c.authentik.ProvidersApi.ProvidersProxyPartialUpdate(ctx, int32(pk)).PatchedProxyProviderRequest(*proxyProviderRequest).Execute()
+	if err != nil {
+		return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyPartialUpdate`: %w with response %v", err, r)
+	}
+
+	pp.Status.PK = strconv.Itoa(int(resp.Pk))
+
+	return c.updateProxyProviderStatus(ctx, pp)
+}
+
+func (c *Controller) reconcileCreate(ctx context.Context, pp *v1.ProxyProvider) error {
+	proxyProviderRequest := &authentikapi.ProxyProviderRequest{
+		Name:              pp.Spec.Name,
+		AuthorizationFlow: pp.Spec.AuthorizationFlow,
+		InvalidationFlow:  pp.Spec.InvalidationFlow,
+		ExternalHost:      pp.Spec.ExternalHost,
+		Mode:              authentikapi.PROXYMODE_FORWARD_SINGLE.Ptr(),
+	}
+	resp, r, err := c.authentik.ProvidersApi.ProvidersProxyCreate(ctx).ProxyProviderRequest(*proxyProviderRequest).Execute()
+	if err != nil {
+		return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyCreate`: %w with response %v", err, r)
+	}
+
+	pp.Status.PK = strconv.Itoa(int(resp.Pk))
+	return c.updateProxyProviderStatus(ctx, pp)
 }
 
 func (c *Controller) enqueueProxyProvider(obj interface{}) {
@@ -248,26 +278,18 @@ func (c *Controller) enqueueProxyProvider(obj interface{}) {
 	c.workqueue.Add(objectRef)
 }
 
-func (c *Controller) handleObject(obj interface{}) {
-	// Optional: resolve Deployment owners back to ProxyProvider and enqueue.
-	_, ok := obj.(metav1.Object)
-	if !ok {
-		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-		if !ok {
-			utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
-			return
-		}
-		_, ok = tombstone.Obj.(metav1.Object)
-		if !ok {
-			utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a metav1.Object %#v", obj))
-			return
-		}
-	}
-}
-
 func (c *Controller) updateProxyProviderStatus(ctx context.Context, pp *v1.ProxyProvider) error {
 	ppCopy := pp.DeepCopy()
-	ppCopy.Status.PK = pp.Status.PK
+	_, err := c.proxyProviderClientset.ProxyproviderV1().ProxyProviders(ppCopy.Namespace).UpdateStatus(ctx, ppCopy, metav1.UpdateOptions{FieldManager: FieldManager})
-	_, err := c.proxyProviderClientset.ProxyproviderV1().ProxyProviders(pp.Namespace).UpdateStatus(ctx, ppCopy, metav1.UpdateOptions{FieldManager: FieldManager})
 	return err
 }
+
+// Update metadata, spec, etc. of the ProxyProvider object.
+func (c *Controller) updateProxyProvider(ctx context.Context, pp *v1.ProxyProvider) error {
+	ppCopy := pp.DeepCopy()
+	_, err := c.proxyProviderClientset.ProxyproviderV1().ProxyProviders(ppCopy.Namespace).Update(ctx, ppCopy, metav1.UpdateOptions{FieldManager: FieldManager})
+	if err != nil {
+		return fmt.Errorf("error updating ProxyProvider metadata: %v", err)
+	}
+	return nil
+}

main.go

@@ -25,7 +25,6 @@ import (
 
 	"gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/signals"
 	authentikapi "goauthentik.io/api/v3"
-	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
@@ -74,16 +73,14 @@ func main() {
 		klog.FlushAndExit(klog.ExitFlushTimeout, 1)
 	}
 
-	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)
 	proxyProviderInformerFactory := informers.NewSharedInformerFactory(proxyProviderClient, time.Second*30)
 
 	controller := NewController(ctx, kubeClient, proxyProviderClient, authentikClient,
-		kubeInformerFactory.Apps().V1().Deployments(),
+		proxyProviderInformerFactory.Proxyprovider().V1().ProxyProviders(),
-		proxyProviderInformerFactory.Proxyprovider().V1().ProxyProviders())
+	)
 
 	// notice that there is no need to run Start methods in a separate goroutine. (i.e. go kubeInformerFactory.Start(ctx.done())
 	// Start method is non-blocking and runs all registered informers in a dedicated goroutine.
-	kubeInformerFactory.Start(ctx.Done())
 	proxyProviderInformerFactory.Start(ctx.Done())
 
 	if err = controller.Run(ctx, 2); err != nil {