refactor: centralize Kubernetes cache startup and synchronization for indexer policies and update deployment restart logic.

Author: JoseSzycho

Taskfile.yaml

@@ -377,6 +377,7 @@ tasks:
         echo "  Etcd:              task test-infra:kubectl -- logs -l app.kubernetes.io/name=etcd -n etcd-system -f"
         echo "  Search API Server:     task test-infra:kubectl -- logs -l app.kubernetes.io/name=search-apiserver -n search-system -f"
         echo "  Search Controller:     task test-infra:kubectl -- logs -l app.kubernetes.io/name=search-controller-manager -n search-system -f"
+        echo "  Search Indexer:     task test-infra:kubectl -- logs -l app.kubernetes.io/name=resource-indexer -n search-system -f"
 
   dev:generate-webhook-certs:
     desc: Generate all certificates for webhook server
@@ -661,7 +662,6 @@ tasks:
     deps:
       - dev:build
       - dev:load
-      - dev:deploy
     cmds:
       - |
         set -e
@@ -672,13 +672,16 @@ tasks:
         # Restart the deployment to pick up new image
         task test-infra:kubectl -- rollout restart deployment/search-apiserver -n search-system
         task test-infra:kubectl -- rollout restart deployment/search-controller-manager -n search-system
+        task test-infra:kubectl -- rollout restart deployment/resource-indexer -n search-system
 
         # Wait for rollout to complete
         echo "Waiting for rollout to complete..."
+        task test-infra:kubectl -- rollout status deployment/search-apiserver -n search-system --timeout=1000s
         task test-infra:kubectl -- rollout status deployment/search-controller-manager -n search-system --timeout=1000s
+        task test-infra:kubectl -- rollout status deployment/resource-indexer -n search-system --timeout=1000s
 
         echo "✅ Redeployment complete!"
-        echo "Check logs with: task test-infra:kubectl -- logs -n search-system -l app.kubernetes.io/name=search-controller-manager"
+        echo "Check pods with: task test-infra:kubectl -- get pods -n search-system"
 
   dev:nats-queue:
     desc: View the NATS queue for the indexer

cmd/search/indexer/command.go

@@ -188,18 +188,28 @@ func Run(o *ResourceIndexerOptions, ctx context.Context) error {
 		return fmt.Errorf("failed to create policy cache: %w", err)
 	}
 
-	go func() {
-		if err := indexPolicyCache.Start(ctx); err != nil {
-			klog.Errorf("Index Policy cache stopped: %v", err)
-		}
-	}()
+	// Register handlers for both caches. They share the same underlying informer.
+	if err := indexPolicyCache.RegisterHandlers(ctx); err != nil {
+		return fmt.Errorf("failed to register index policy handlers: %w", err)
+	}
+	if err := reindexPolicyCache.RegisterHandlers(ctx); err != nil {
+		return fmt.Errorf("failed to register reindex policy handlers: %w", err)
+	}
 
+	// Start the shared cache and wait for it to be synced.
 	go func() {
-		if err := reindexPolicyCache.Start(ctx); err != nil {
-			klog.Errorf("Reindex Policy cache stopped: %v", err)
+		klog.Info("Starting shared Kubernetes cache...")
+		if err := k8sCache.Start(ctx); err != nil {
+			klog.Fatalf("Kubernetes cache stopped with error: %v", err)
 		}
 	}()
 
+	klog.Info("Waiting for cache to sync...")
+	if !k8sCache.WaitForCacheSync(ctx) {
+		return fmt.Errorf("failed to sync Kubernetes cache")
+	}
+	klog.Info("Cache synced successfully")
+
 	// Connect to NATS
 	klog.Infof("Connecting to NATS at %s...", o.NatsURL)
 	nc, err := nats.Connect(o.NatsURL)

internal/indexer/policy_cache.go

@@ -53,9 +53,9 @@ func NewPolicyCache(c runtimecache.Cache, requireReadyCondition bool) (*PolicyCa
 	}, nil
 }
 
-// Start registers informer event handlers for ResourceIndexPolicy objects
-func (c *PolicyCache) Start(ctx context.Context) error {
-	klog.Info("Starting policy cache informer")
+// RegisterHandlers registers informer event handlers for ResourceIndexPolicy objects.
+func (c *PolicyCache) RegisterHandlers(ctx context.Context) error {
+	klog.Info("Registering policy cache informer handlers")
 
 	informer, err := c.cache.GetInformer(ctx, &v1alpha1.ResourceIndexPolicy{})
 	if err != nil {
@@ -95,10 +95,6 @@ func (c *PolicyCache) Start(ctx context.Context) error {
 		return fmt.Errorf("failed to add event handler to ResourceIndexPolicy informer: %w", err)
 	}
 
-	// Start runs all informers and blocks until ctx is cancelled.
-	if err := c.cache.Start(ctx); err != nil {
-		return fmt.Errorf("policy cache informer stopped with error: %w", err)
-	}
 	return nil
 }
 
@@ -112,7 +108,7 @@ func (c *PolicyCache) upsertPolicy(p *v1alpha1.ResourceIndexPolicy) {
 	// still being initialized (e.g. index creation or initial re-indexing).
 	if c.requireReadyCondition {
 		if !meta.IsStatusConditionTrue(p.Status.Conditions, "Ready") {
-			klog.Infof("Policy %s is not yet Ready; skipping cache", key)
+			klog.Infof("Policy %s is not yet Ready (Ready=True condition missing); skipping cache", key)
 			c.deletePolicy(key)
 			return
 		}

internal/indexer/reindex_consumer.go

@@ -64,6 +64,7 @@ func (r *ReindexConsumer) Start(ctx context.Context) error {
 		for _, cp := range policies {
 			// Skip if index name is not set yet
 			if cp.Policy.Status.IndexName == "" {
+				klog.V(2).Infof("ReindexConsumer: policy %s has no IndexName in status, skipping", cp.Policy.Name)
 				continue
 			}
 
@@ -86,6 +87,8 @@ func (r *ReindexConsumer) Start(ctx context.Context) error {
 				r.batcher.QueueUpsert(cp.Policy.Status.IndexName, doc, &msg)
 				queued = true
 			} else {
+				klog.V(4).Infof("ReindexConsumer: policy %s did not match resource %s/%s (id=%s), ensuring deletion", cp.Policy.Name, obj.GetNamespace(), obj.GetName(), event.ID)
+
 				// If it doesn't match this policy, we should ensure it's removed from the index
 				// in case it was previously indexed there.
 				r.batcher.QueueDelete(cp.Policy.Status.IndexName, resourceUID, &msg)