feat: enable multi-tenant indexing with a new configuration flag and resource filtering based on tenant type.

Author: JoseSzycho

cmd/search/indexer/command.go

@@ -302,7 +302,7 @@ func Run(o *ResourceIndexerOptions, ctx context.Context) error {
 	auditBatcher.Start(ctx)
 	reindexBatcher.Start(ctx)
 
-	auditIdx := indexer.NewIndexer(auditConsumer, indexPolicyCache, auditBatcher)
+	auditIdx := indexer.NewIndexer(auditConsumer, indexPolicyCache, auditBatcher, o.MultiTenant)
 	reindexIdx := indexer.NewReindexConsumer(reindexJSConsumer, reindexPolicyCache, reindexBatcher)
 
 	klog.Info("Starting audit indexer and re-index consumer...")

config/base/resource-indexer/deployment.yaml

@@ -28,6 +28,7 @@ spec:
         - --nats-tls-cert=$(NATS_TLS_CERT)
         - --nats-tls-key=$(NATS_TLS_KEY)
         - --meilisearch-domain=$(MEILISEARCH_DOMAIN)
+        - --multi-tenant=$(MULTI_TENANT)
         env:
         - name: NATS_URL
           value: "nats://nats.nats-system.svc.cluster.local:4222"
@@ -47,6 +48,8 @@ spec:
           value: "AUDIT_EVENTS"
         - name: MEILISEARCH_DOMAIN
           value: "http://meilisearch.meilisearch-system.svc.cluster.local:7700"
+        - name: MULTI_TENANT
+          value: "false"
         - name: MEILISEARCH_API_KEY
           valueFrom:
             secretKeyRef:

config/overlays/resource-indexer/core-control-plane/patches/deployment.yaml

@@ -7,3 +7,8 @@ spec:
     spec:
       serviceAccountName: resource-indexer
       automountServiceAccountToken: true
+      containers:
+      - name: indexer
+        env:
+        - name: MULTI_TENANT
+          value: "true"

internal/indexer/consumer.go

@@ -16,6 +16,7 @@ type Indexer struct {
 	consumer    jetstream.Consumer
 	policyCache *PolicyCache
 	batcher     *Batcher
+	multiTenant bool
 	mu          sync.Mutex
 }
 
@@ -59,17 +60,24 @@ func extractTenantFromAuditEvent(event *auditEvent) (tenantName string, tenantTy
 }
 
 // NewIndexer creates a new Indexer instance.
-func NewIndexer(consumer jetstream.Consumer, policyCache *PolicyCache, batcher *Batcher) *Indexer {
+func NewIndexer(consumer jetstream.Consumer, policyCache *PolicyCache, batcher *Batcher, multiTenant bool) *Indexer {
 	return &Indexer{
 		consumer:    consumer,
 		policyCache: policyCache,
 		batcher:     batcher,
+		multiTenant: multiTenant,
 	}
 }
 
 var upsertVerbs = map[string]bool{"create": true, "update": true, "patch": true}
 
-const deleteVerb = "delete"
+const (
+	deleteVerb = "delete"
+	// tenantTypePlatform mirrors tenant.TenantTypePlatform. A local copy is used
+	// to avoid an import cycle: internal/tenant/project_watcher.go already
+	// imports internal/indexer, so internal/indexer cannot import internal/tenant.
+	tenantTypePlatform = "platform"
+)
 
 // Start starts the indexer consumer loop.
 // Note: the Batcher must be started separately by the caller (via batcher.Start)
@@ -147,8 +155,14 @@ func (i *Indexer) Start(ctx context.Context) error {
 					continue
 				}
 
-				// Inject tenant context extracted from the audit event's user extra fields.
+				// Always extract tenant context from the audit event's user extra fields.
+				// In single-tenant mode, skip non-platform resources so that project-tenant
+				// resources are never incorrectly indexed as platform resources.
 				evalResult.Tenant, evalResult.TenantType = extractTenantFromAuditEvent(&event)
+				if !i.multiTenant && evalResult.TenantType != tenantTypePlatform {
+					msg.Ack()
+					continue
+				}
 
 				// Transform the matching resource into an indexable document
 				doc := evalResult.Transform()

internal/indexer/consumer_test.go

@@ -136,7 +136,7 @@ func TestIndexer_Start_ConsumeFlow(t *testing.T) {
 	mockSearch.On("WaitForTasks", mock.Anything).Return(nil, nil).Once()
 
 	// 4. Run Indexer
-	indexer := NewIndexer(mockConsumer, policyCache, batcher)
+	indexer := NewIndexer(mockConsumer, policyCache, batcher, false)
 	ctx, cancel := context.WithCancel(context.Background())
 
 	// Run Start in a goroutine
@@ -275,7 +275,7 @@ func TestIndexer_Consume_Delete(t *testing.T) {
 	})).Return(nil, nil).Once()
 	mockSearch.On("WaitForTasks", mock.Anything).Return(nil, nil).Once()
 
-	indexer := NewIndexer(mockConsumer, policyCache, batcher)
+	indexer := NewIndexer(mockConsumer, policyCache, batcher, false)
 	ctx, cancel := context.WithCancel(context.Background())
 
 	go indexer.Start(ctx)