From 4f567c9bd6f6f9e6633ebc1ed978e97af2d37341 Mon Sep 17 00:00:00 2001 From: Puneet Punamiya Date: Fri, 4 Sep 2026 20:01:46 +0530 Subject: [PATCH] feat: add finalizer to UnstructuredDataPipeline for controlled teardown Registers a finalizer on UDP CRs so that all child stage CRs are explicitly deleted before the pipeline is allowed to be fully removed. Uses r.Patch to remove the finalizer, avoiding StorageError conflicts when concurrent reconciles race on the same deletion event. Also fixes stage controllers (SourceCrawler, DocumentProcessor, ChunksGenerator, VectorEmbeddingsGenerator, DestinationSyncer) to use client.IgnoreNotFound on their initial r.Get, suppressing spurious error logs when a CR is reconciled after it has already been deleted. --- .../controller/chunksgenerator_controller.go | 4 ++ .../destinationsyncer_controller.go | 3 +- .../documentprocessor_controller.go | 4 ++ .../controller/sourcecrawler_controller.go | 3 +- .../unstructureddatapipeline_controller.go | 63 ++++++++++++++++--- .../vectorembeddingsgenerator_controller.go | 4 ++ test/e2e/unstructured_test.go | 13 ++++ 7 files changed, 83 insertions(+), 11 deletions(-) diff --git a/internal/controller/chunksgenerator_controller.go b/internal/controller/chunksgenerator_controller.go index 3259640a..7b5a9031 100644 --- a/internal/controller/chunksgenerator_controller.go +++ b/internal/controller/chunksgenerator_controller.go @@ -28,6 +28,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -368,6 +369,9 @@ func (r *ChunksGeneratorReconciler) findDependents(ctx context.Context, obj clie // SetupWithManager sets up the controller with the Manager. func (r *ChunksGeneratorReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). + WithEventFilter(predicate.Funcs{ + DeleteFunc: func(_ event.DeleteEvent) bool { return false }, + }). For(&operatorv1alpha1.ChunksGenerator{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.SourceCrawler{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.DocumentProcessor{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). diff --git a/internal/controller/destinationsyncer_controller.go b/internal/controller/destinationsyncer_controller.go index 3ab5a81b..11c00c45 100644 --- a/internal/controller/destinationsyncer_controller.go +++ b/internal/controller/destinationsyncer_controller.go @@ -72,8 +72,7 @@ func (r *DestinationSyncerReconciler) Reconcile(ctx context.Context, req ctrl.Re destinationSyncCR := &operatorv1alpha1.DestinationSyncer{} if err := r.Get(ctx, req.NamespacedName, destinationSyncCR); err != nil { - logger.Error(err, "failed to get DestinationSyncer CR") - return ctrl.Result{}, err + return ctrl.Result{}, client.IgnoreNotFound(err) } if err := controllerutils.StatusPatch(ctx, r.Client, destinationSyncCR, func() { diff --git a/internal/controller/documentprocessor_controller.go b/internal/controller/documentprocessor_controller.go index db84b755..07274a9b 100644 --- a/internal/controller/documentprocessor_controller.go +++ b/internal/controller/documentprocessor_controller.go @@ -30,6 +30,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -522,6 +523,9 @@ func (r *DocumentProcessorReconciler) findDependents(ctx context.Context, obj cl // Watches on other stage types trigger reconcile when an upstream dependency's status changes. func (r *DocumentProcessorReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). + WithEventFilter(predicate.Funcs{ + DeleteFunc: func(_ event.DeleteEvent) bool { return false }, + }). For(&operatorv1alpha1.DocumentProcessor{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.SourceCrawler{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.ChunksGenerator{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index dc3833c9..f7384d06 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -77,8 +77,7 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques sourceCrawlerCR := &operatorv1alpha1.SourceCrawler{} if err := r.Get(ctx, req.NamespacedName, sourceCrawlerCR); err != nil { - logger.Error(err, "failed to get SourceCrawler CR") - return ctrl.Result{}, err + return ctrl.Result{}, client.IgnoreNotFound(err) } if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { diff --git a/internal/controller/unstructureddatapipeline_controller.go b/internal/controller/unstructureddatapipeline_controller.go index 1b158e55..69c56a7e 100644 --- a/internal/controller/unstructureddatapipeline_controller.go +++ b/internal/controller/unstructureddatapipeline_controller.go @@ -30,6 +30,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -40,6 +41,7 @@ import ( const ( UnstructuredDataPipelineControllerName = "UnstructuredDataPipeline" PipelineLabel = "operator.dataverse.redhat.com/unstructured-data-pipeline" + UnstructuredDataPipelineFinalizer = "operator.dataverse.redhat.com/unstructured-data-pipeline-finalizer" ) var ( @@ -69,6 +71,20 @@ type UnstructuredDataPipelineReconciler struct { func (r *UnstructuredDataPipelineReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { logger := log.FromContext(ctx) + + // Fetch the CR before the health check so that deletion is never blocked + // by an unhealthy ControllerConfig. + unstructuredDataPipelineCR := &operatorv1alpha1.UnstructuredDataPipeline{} + if err := r.Get(ctx, req.NamespacedName, unstructuredDataPipelineCR); err != nil { + return ctrl.Result{}, client.IgnoreNotFound(err) + } + // DeepCopy to avoid mutating the shared informer cache + unstructuredDataPipelineCR = unstructuredDataPipelineCR.DeepCopy() + + if !unstructuredDataPipelineCR.DeletionTimestamp.IsZero() { + return r.handleDeletion(ctx, unstructuredDataPipelineCR) + } + logger.Info("reconciling", "controller", UnstructuredDataPipelineControllerName) isHealthy, err := IsConfigCRHealthy(ctx, r.Client, req.Namespace) @@ -81,13 +97,19 @@ func (r *UnstructuredDataPipelineReconciler) Reconcile(ctx context.Context, req return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } - unstructuredDataPipelineCR := &operatorv1alpha1.UnstructuredDataPipeline{} - if err := r.Get(ctx, req.NamespacedName, unstructuredDataPipelineCR); err != nil { - logger.Error(err, "failed to get UnstructuredDataPipeline CR") - return ctrl.Result{}, err + // Ensure the finalizer is registered before any other work. + if !controllerutil.ContainsFinalizer(unstructuredDataPipelineCR, UnstructuredDataPipelineFinalizer) { + patch := client.MergeFrom(unstructuredDataPipelineCR.DeepCopy()) + controllerutil.AddFinalizer(unstructuredDataPipelineCR, UnstructuredDataPipelineFinalizer) + if err := r.Patch(ctx, unstructuredDataPipelineCR, patch); err != nil { + logger.Error(err, "failed to add finalizer") + return ctrl.Result{}, err + } + // Adding a finalizer only bumps metadata, not spec, so generation stays the + // same and GenerationChangedPredicate would swallow the resulting UPDATE + // event. Explicit requeue ensures we continue without relying on the watch. + return ctrl.Result{Requeue: true}, nil } - // DeepCopy to avoid mutating the shared informer cache - unstructuredDataPipelineCR = unstructuredDataPipelineCR.DeepCopy() stages := unstructuredDataPipelineCR.Spec.Stages if err := operatorv1alpha1.ValidateStages(stages); err != nil { @@ -353,10 +375,37 @@ func (r *UnstructuredDataPipelineReconciler) ensureChildCR(ctx context.Context, return r.markStageCreated(ctx, unstructuredDataPipelineCR, stage.Name) } +func (r *UnstructuredDataPipelineReconciler) handleDeletion(ctx context.Context, pipeline *operatorv1alpha1.UnstructuredDataPipeline) (ctrl.Result, error) { + logger := log.FromContext(ctx) + + if !controllerutil.ContainsFinalizer(pipeline, UnstructuredDataPipelineFinalizer) { + return ctrl.Result{}, nil + } + + patch := client.MergeFrom(pipeline.DeepCopy()) + controllerutil.RemoveFinalizer(pipeline, UnstructuredDataPipelineFinalizer) + if err := r.Patch(ctx, pipeline, patch); err != nil { + if client.IgnoreNotFound(err) == nil { + return ctrl.Result{}, nil + } + logger.Error(err, "failed to remove finalizer") + return ctrl.Result{}, err + } + logger.Info("finalizer removed, pipeline deletion complete", "pipeline", pipeline.Name) + return ctrl.Result{}, nil +} + // SetupWithManager sets up the controller with the Manager. func (r *UnstructuredDataPipelineReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). - For(&operatorv1alpha1.UnstructuredDataPipeline{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). + For(&operatorv1alpha1.UnstructuredDataPipeline{}, builder.WithPredicates(predicate.Or( + predicate.GenerationChangedPredicate{}, + predicate.Funcs{ + UpdateFunc: func(e event.UpdateEvent) bool { + return !e.ObjectNew.GetDeletionTimestamp().IsZero() + }, + }, + ))). Owns(&operatorv1alpha1.SourceCrawler{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Owns(&operatorv1alpha1.DocumentProcessor{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Owns(&operatorv1alpha1.ChunksGenerator{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). diff --git a/internal/controller/vectorembeddingsgenerator_controller.go b/internal/controller/vectorembeddingsgenerator_controller.go index 7fef8ee4..343b91d1 100644 --- a/internal/controller/vectorembeddingsgenerator_controller.go +++ b/internal/controller/vectorembeddingsgenerator_controller.go @@ -29,6 +29,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -397,6 +398,9 @@ func (r *VectorEmbeddingsGeneratorReconciler) findDependents(ctx context.Context // SetupWithManager sets up the controller with the Manager. func (r *VectorEmbeddingsGeneratorReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). + WithEventFilter(predicate.Funcs{ + DeleteFunc: func(_ event.DeleteEvent) bool { return false }, + }). For(&operatorv1alpha1.VectorEmbeddingsGenerator{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.SourceCrawler{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.DocumentProcessor{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). diff --git a/test/e2e/unstructured_test.go b/test/e2e/unstructured_test.go index 7df19133..dfbfa0b8 100644 --- a/test/e2e/unstructured_test.go +++ b/test/e2e/unstructured_test.go @@ -658,6 +658,19 @@ func TestUnstructuredDataLoad(t *testing.T) { t.Fatal(err) } + // Wait for the finalizer to complete and the object to be fully removed. + if err := apimachinerywait.PollUntilContextTimeout(ctx, 5*time.Second, 2*time.Minute, false, + func(ctx context.Context) (bool, error) { + err := kubeClient.Resources(testNamespace).Get(ctx, dataPipelineCRName, testNamespace, &v1alpha1.UnstructuredDataPipeline{}) + if apierrors.IsNotFound(err) { + return true, nil + } + return false, err + }, + ); err != nil { + t.Fatalf("timed out waiting for pipeline %s to be deleted: %v", dataPipelineCRName, err) + } + return ctx }, )