feat: Implement ResourceIndexPolicy finalizer to clean up Meilisearch indexes and update policy cache eviction logic.

Author: JoseSzycho

config/overlays/controller-manager/core-control-plane/rbac/role.yaml

@@ -19,7 +19,15 @@ rules:
   verbs:
   - get
   - list
+  - patch
+  - update
   - watch
+- apiGroups:
+  - search.miloapis.com
+  resources:
+  - resourceindexpolicies/finalizers
+  verbs:
+  - update
 - apiGroups:
   - search.miloapis.com
   resources:

internal/controllers/policy/policy_controller.go

@@ -19,6 +19,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/finalizer"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	"go.miloapis.net/search/internal/cel"
@@ -57,12 +58,17 @@ type ResourceIndexPolicyReconciler struct {
 
 	RetryBaseDelay time.Duration
 	RetryMaxDelay  time.Duration
+
+	Finalizers finalizer.Finalizers
 }
 
 const (
-	ReadyConditionType      = "Ready"
-	ReadyConditionReason    = "PolicyReady"
-	NotReadyConditionReason = "PolicyNotReady"
+	FinalizerName = "search.miloapis.com/cleanup"
+
+	ReadyConditionType         = "Ready"
+	ReadyConditionReason       = "PolicyReady"
+	NotReadyConditionReason    = "PolicyNotReady"
+	TerminatingConditionReason = "Terminating"
 
 	SearchIndexReadyConditionType = "SearchIndexReady"
 	IndexCreatedReason            = "IndexCreated"
@@ -83,8 +89,9 @@ const (
 	ReindexingInProgressReason = "ReindexingInProgress"
 )
 
-// +kubebuilder:rbac:groups=search.miloapis.com,resources=resourceindexpolicies,verbs=get;list;watch
+// +kubebuilder:rbac:groups=search.miloapis.com,resources=resourceindexpolicies,verbs=get;list;watch;update;patch
 // +kubebuilder:rbac:groups=search.miloapis.com,resources=resourceindexpolicies/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups=search.miloapis.com,resources=resourceindexpolicies/finalizers,verbs=update
 // +kubebuilder:rbac:groups=*,resources=*,verbs=get;list;watch
 
 // Reconcile matches the state of the cluster with the desired state of a ResourceIndexPolicy.
@@ -104,9 +111,26 @@ func (r *ResourceIndexPolicyReconciler) Reconcile(ctx context.Context, req ctrl.
 		return ctrl.Result{}, err
 	}
 
-	// Check if the policy is being deleted
+	// Run finalizers:
+	finalizeResult, err := r.Finalizers.Finalize(ctx, policy)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed to run finalizers: %w", err)
+	}
+	if finalizeResult.Updated {
+		logger.Info("Finalizer updated the policy object, persisting to API server")
+		if updateErr := r.Client.Update(ctx, policy); updateErr != nil {
+			if errors.IsConflict(updateErr) {
+				logger.Info("Conflict updating policy after finalizer update; requeuing")
+				return ctrl.Result{Requeue: true}, nil
+			}
+			logger.Error(updateErr, "Failed to update ResourceIndexPolicy after finalizer update")
+			return ctrl.Result{}, updateErr
+		}
+		return ctrl.Result{}, nil
+	}
+
 	if policy.GetDeletionTimestamp() != nil {
-		logger.Info("ResourceIndexPolicy is being deleted")
+		logger.Info("ResourceIndexPolicy is being deleted, skipping reconciliation")
 		return ctrl.Result{}, nil
 	}
 
@@ -493,8 +517,67 @@ func (r *ResourceIndexPolicyReconciler) publishReindexMessages(
 	return nil
 }
 
+// resourceIndexPolicyFinalizer handles Meilisearch cleanup when a
+// ResourceIndexPolicy is deleted.
+type resourceIndexPolicyFinalizer struct {
+	Client    client.Client
+	SearchSDK *meilisearch.SDKClient
+}
+
+func (f *resourceIndexPolicyFinalizer) Finalize(ctx context.Context, obj client.Object) (finalizer.Result, error) {
+	log := logf.FromContext(ctx).WithName("resourceindexpolicy-finalizer")
+	log.Info("Finalizing ResourceIndexPolicy")
+
+	policy, ok := obj.(*searchv1alpha1.ResourceIndexPolicy)
+	if !ok {
+		return finalizer.Result{}, fmt.Errorf("object is not a ResourceIndexPolicy")
+	}
+
+	// Set Ready=False immediately so consumers can observe the terminating state.
+	oldStatus := policy.Status.DeepCopy()
+	meta.SetStatusCondition(&policy.Status.Conditions, metav1.Condition{
+		Type:    ReadyConditionType,
+		Status:  metav1.ConditionFalse,
+		Reason:  TerminatingConditionReason,
+		Message: "ResourceIndexPolicy is being deleted",
+	})
+	if err := utils.UpdateStatusIfChanged(ctx, f.Client, log, policy, oldStatus, &policy.Status); err != nil {
+		log.Error(err, "Failed to update ResourceIndexPolicy status")
+		return finalizer.Result{}, err
+	}
+
+	// Remove all documents then delete the index itself.
+	searchIndex := policy.Status.IndexName
+	if searchIndex == "" {
+		log.Info("No index name found, skipping cleanup")
+		return finalizer.Result{}, nil
+	}
+	log.Info("Deleting all documents from search index", "index", searchIndex)
+	if err := f.SearchSDK.DeleteAllDocuments(searchIndex); err != nil {
+		log.Error(err, "Failed to delete all documents from search index")
+		return finalizer.Result{}, err
+	}
+
+	log.Info("Deleting search index", "index", searchIndex)
+	if err := f.SearchSDK.DeleteIndex(searchIndex); err != nil {
+		log.Error(err, "Failed to delete search index")
+		return finalizer.Result{}, err
+	}
+
+	log.Info("ResourceIndexPolicy cleanup complete")
+	return finalizer.Result{}, nil
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *ResourceIndexPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	r.Finalizers = finalizer.NewFinalizers()
+	if err := r.Finalizers.Register(FinalizerName, &resourceIndexPolicyFinalizer{
+		Client:    r.Client,
+		SearchSDK: r.SearchSDK,
+	}); err != nil {
+		return fmt.Errorf("failed to register finalizer: %w", err)
+	}
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&searchv1alpha1.ResourceIndexPolicy{}).
 		Complete(r)

internal/indexer/policy_cache.go

@@ -103,6 +103,16 @@ func (c *PolicyCache) RegisterHandlers(ctx context.Context) error {
 func (c *PolicyCache) upsertPolicy(p *v1alpha1.ResourceIndexPolicy) {
 	key := p.Name
 
+	// Evict policies that are being deleted so the indexer stops processing new
+	// resources against an index that is about to be torn down. DeletionTimestamp
+	// is set by Kubernetes as soon as a delete request is received — before the
+	// finalizer runs — giving us the earliest possible eviction signal.
+	if p.DeletionTimestamp != nil {
+		klog.Infof("Policy %s is being deleted; evicting from cache", key)
+		c.deletePolicy(key)
+		return
+	}
+
 	// If strict ready checking is enabled, we only cache policies that are fully Ready.
 	// This prevents the primary indexer from processing events for policies that are
 	// still being initialized (e.g. index creation or initial re-indexing).

pkg/meilisearch/sdk_client.go

@@ -337,6 +337,56 @@ func (s *SDKClient) UpdateSearchableAttributes(indexUID string, attributes []str
 	return &meilisearch.Task{TaskUID: resp.TaskUID, IndexUID: indexUID}, nil
 }
 
+// DeleteAllDocuments deletes all documents from the given index and waits for the
+// task to complete before returning. If the index does not exist, it is a no-op.
+func (s *SDKClient) DeleteAllDocuments(indexUID string) error {
+	resp, err := s.client.Index(indexUID).DeleteAllDocuments(nil)
+	if err != nil {
+		if strings.Contains(err.Error(), "index_not_found") {
+			klog.Infof("Index %s not found, skipping document deletion", indexUID)
+			return nil
+		}
+		return fmt.Errorf("failed to delete all documents from index %s: %w", indexUID, err)
+	}
+
+	task, err := s.waitForTask(resp.TaskUID)
+	if err != nil {
+		return fmt.Errorf("failed to wait for document deletion task: %w", err)
+	}
+
+	if task.Status != meilisearch.TaskStatusSucceeded {
+		return fmt.Errorf("document deletion task failed with status %s: %s", task.Status, task.Error.Message)
+	}
+
+	klog.Infof("All documents deleted from index %s", indexUID)
+	return nil
+}
+
+// DeleteIndex deletes the index with the given UID and waits for the task to
+// complete. If the index does not exist, it is a no-op.
+func (s *SDKClient) DeleteIndex(indexUID string) error {
+	resp, err := s.client.DeleteIndex(indexUID)
+	if err != nil {
+		if strings.Contains(err.Error(), "index_not_found") {
+			klog.Infof("Index %s not found, skipping deletion", indexUID)
+			return nil
+		}
+		return fmt.Errorf("failed to delete index %s: %w", indexUID, err)
+	}
+
+	task, err := s.waitForTask(resp.TaskUID)
+	if err != nil {
+		return fmt.Errorf("failed to wait for index deletion task: %w", err)
+	}
+
+	if task.Status != meilisearch.TaskStatusSucceeded {
+		return fmt.Errorf("index deletion task failed with status %s: %s", task.Status, task.Error.Message)
+	}
+
+	klog.Infof("Index %s deleted", indexUID)
+	return nil
+}
+
 // GetSettingsUpdateTask returns the most recent settings update task for the given index.
 func (s *SDKClient) GetSettingsUpdateTask(indexUID string) (*meilisearch.Task, error) {
 	resp, err := s.client.GetTasks(&meilisearch.TasksQuery{