feat: update policycache to use informer

Author: JoseSzycho

cmd/search/indexer/command.go

@@ -15,7 +15,7 @@ import (
 	"go.miloapis.net/search/pkg/meilisearch"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/klog/v2"
-	"sigs.k8s.io/controller-runtime/pkg/client"
+	runtimecache "sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 )
 
@@ -30,11 +30,6 @@ type ResourceIndexerOptions struct {
 	NatsAckWait     time.Duration
 	NatsMaxInFlight int
 
-	// ResourceIndexPolicySyncInterval controls how often the indexer polls the Kubernetes API
-	// to refresh the cache of ResourceIndexPolicies. This allows the indexer to pick up
-	// changes to indexing rules without a restart.
-	ResourceIndexPolicySyncInterval time.Duration
-
 	// Meilisearch connection and timeout settings
 	MeilisearchTaskWaitTimeout time.Duration // Timeout for waiting for Meilisearch tasks to complete.
 	MeilisearchHTTPTimeout     time.Duration // Timeout for HTTP requests to Meilisearch.
@@ -52,23 +47,22 @@ type ResourceIndexerOptions struct {
 // NewResourceIndexerOptions creates a new ResourceIndexerOptions with default values.
 func NewResourceIndexerOptions() *ResourceIndexerOptions {
 	return &ResourceIndexerOptions{
-		NatsURL:                         "nats://nats.nats-system.svc.cluster.local:4222",
-		NatsSubject:                     "audit.>",
-		NatsQueueGroup:                  "search-indexer",
-		NatsDurableName:                 "search-indexer",
-		NatsStreamName:                  "AUDIT_EVENTS",
-		NatsAckWait:                     120 * time.Second,
-		NatsMaxInFlight:                 10000,
-		ResourceIndexPolicySyncInterval: 2 * time.Minute,
-		MeilisearchTaskWaitTimeout:      4 * time.Second,
-		MeilisearchHTTPTimeout:          60 * time.Second,
-		MeilisearchDomain:               "http://meilisearch.meilisearch-system.svc.cluster.local:7700",
-		MeilisearchChunkSize:            1000,
-		BatchSize:                       1000,
-		FlushInterval:                   5 * time.Second,
-		MeilisearchMaxRetries:           3,
-		MeilisearchRetryDelay:           500 * time.Millisecond,
-		BatchMaxConcurrentUploads:       100,
+		NatsURL:                    "nats://nats.nats-system.svc.cluster.local:4222",
+		NatsSubject:                "audit.>",
+		NatsQueueGroup:             "search-indexer",
+		NatsDurableName:            "search-indexer",
+		NatsStreamName:             "AUDIT_EVENTS",
+		NatsAckWait:                120 * time.Second,
+		NatsMaxInFlight:            10000,
+		MeilisearchTaskWaitTimeout: 4 * time.Second,
+		MeilisearchHTTPTimeout:     60 * time.Second,
+		MeilisearchDomain:          "http://meilisearch.meilisearch-system.svc.cluster.local:7700",
+		MeilisearchChunkSize:       1000,
+		BatchSize:                  1000,
+		FlushInterval:              5 * time.Second,
+		MeilisearchMaxRetries:      3,
+		MeilisearchRetryDelay:      500 * time.Millisecond,
+		BatchMaxConcurrentUploads:  100,
 	}
 }
 
@@ -81,7 +75,6 @@ func (o *ResourceIndexerOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.NatsStreamName, "nats-stream-name", o.NatsStreamName, "The name of the JetStream stream.")
 	fs.DurationVar(&o.NatsAckWait, "nats-ack-wait", o.NatsAckWait, "The time to wait for an acknowledgement.")
 	fs.IntVar(&o.NatsMaxInFlight, "nats-max-in-flight", o.NatsMaxInFlight, "The maximum number of in-flight messages.")
-	fs.DurationVar(&o.ResourceIndexPolicySyncInterval, "resource-index-policy-sync-interval", o.ResourceIndexPolicySyncInterval, "How often to re-sync ResourceIndexPolicies.")
 	fs.StringVar(&o.MeilisearchDomain, "meilisearch-domain", o.MeilisearchDomain, "Domain of the Meilisearch instance.")
 	fs.DurationVar(&o.MeilisearchTaskWaitTimeout, "meilisearch-task-wait-timeout", o.MeilisearchTaskWaitTimeout, "Timeout for waiting for Meilisearch tasks to complete.")
 	fs.DurationVar(&o.MeilisearchHTTPTimeout, "meilisearch-http-timeout", o.MeilisearchHTTPTimeout, "Timeout for HTTP requests to Meilisearch.")
@@ -116,9 +109,6 @@ func (o *ResourceIndexerOptions) Validate() error {
 	if o.NatsMaxInFlight < 1 {
 		return fmt.Errorf("nats-max-in-flight must be greater than 0")
 	}
-	if o.ResourceIndexPolicySyncInterval < 10*time.Second {
-		return fmt.Errorf("resource-index-policy-sync-interval must be at least 10s")
-	}
 	if o.MeilisearchDomain == "" {
 		return fmt.Errorf("meilisearch-domain must be set")
 	}
@@ -177,7 +167,7 @@ func NewIndexerCommand() *cobra.Command {
 
 // Run starts the indexer consumer
 func Run(o *ResourceIndexerOptions, ctx context.Context) error {
-	// Build a Kubernetes client for listing policies
+	// Build a scheme and REST config for the controller-runtime cache.
 	scheme := runtime.NewScheme()
 	if err := searchv1alpha1.AddToScheme(scheme); err != nil {
 		return fmt.Errorf("failed to add v1alpha1 scheme: %w", err)
@@ -188,13 +178,15 @@ func Run(o *ResourceIndexerOptions, ctx context.Context) error {
 		return fmt.Errorf("failed to get kubeconfig: %w", err)
 	}
 
-	k8sClient, err := client.New(cfg, client.Options{Scheme: scheme})
+	// Create a controller-runtime cache that uses a watch stream (informer)
+	// to keep ResourceIndexPolicies in-sync.
+	k8sCache, err := runtimecache.New(cfg, runtimecache.Options{Scheme: scheme})
 	if err != nil {
-		return fmt.Errorf("failed to create kubernetes client: %w", err)
+		return fmt.Errorf("failed to create controller-runtime cache: %w", err)
 	}
 
 	// Create and start the policy cache
-	policyCache, err := indexer.NewPolicyCache(k8sClient, o.ResourceIndexPolicySyncInterval)
+	policyCache, err := indexer.NewPolicyCache(k8sCache)
 	if err != nil {
 		return fmt.Errorf("failed to create policy cache: %w", err)
 	}

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

@@ -0,0 +1,22 @@
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: milo-controller-manager
+rules:
+- apiGroups:
+  - search.miloapis.com
+  resources:
+  - resourceindexpolicies
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - search.miloapis.com
+  resources:
+  - resourceindexpolicies/status
+  verbs:
+  - get
+  - patch
+  - update

config/samples/policy_v1alpha1_configmap_policy.yaml

@@ -1,4 +1,4 @@
-apiVersion: policy.search.miloapis.com/v1alpha1
+apiVersion: search.miloapis.com/v1alpha1
 kind: ResourceIndexPolicy
 metadata:
   name: configmap-index-policy-1

config/samples/policy_v1alpha1_service_policy.yaml

@@ -1,4 +1,4 @@
-apiVersion: policy.search.miloapis.com/v1alpha1
+apiVersion: search.miloapis.com/v1alpha1
 kind: ResourceIndexPolicy
 metadata:
   name: service-index-policy-1

internal/indexer/batcher.go

@@ -110,12 +110,8 @@ func (b *Batcher) QueueUpsert(indexUID string, doc any, msg *jetstream.Msg) {
 	}
 }
 
-var a int = 0
-
 // QueueDelete adds a document ID to the pending map and triggers an asynchronous flush if the batch size is reached.
 func (b *Batcher) QueueDelete(indexUID string, docID string, msg *jetstream.Msg) {
-	a++
-	klog.Infof("QueueDelete %d, %d", len(b.pendingDeletes), len(b.deleteMsgs))
 	b.mu.Lock()
 	defer b.mu.Unlock()
 

internal/indexer/consumer_test.go

@@ -9,10 +9,10 @@ import (
 
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/mock"
+	internalcel "go.miloapis.net/search/internal/cel"
+	policyevaluation "go.miloapis.net/search/internal/policy/evaluation"
 	"go.miloapis.net/search/pkg/apis/search/v1alpha1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 // MockConsumer is a mock for jetstream.Consumer
@@ -74,9 +74,6 @@ func (m *MockConsumeContext) Closed() <-chan struct{} { return nil }
 
 func TestIndexer_Start_ConsumeFlow(t *testing.T) {
 	// 1. Setup PolicyCache with a policy
-	scheme := runtime.NewScheme()
-	v1alpha1.AddToScheme(scheme)
-
 	policy := &v1alpha1.ResourceIndexPolicy{
 		ObjectMeta: metav1.ObjectMeta{Name: "pod-policy"},
 		Spec: v1alpha1.ResourceIndexPolicySpec{
@@ -91,9 +88,12 @@ func TestIndexer_Start_ConsumeFlow(t *testing.T) {
 		},
 	}
 
-	k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(policy).Build()
-	policyCache, _ := NewPolicyCache(k8sClient, 1*time.Minute)
-	policyCache.refresh(context.Background())
+	env, _ := internalcel.NewEnv()
+	policyCache := &PolicyCache{
+		policies: make(map[string]*policyevaluation.CachedPolicy),
+		celEnv:   env,
+	}
+	policyCache.upsertPolicy(policy)
 
 	// 2. Setup Batcher with Mock Search Client
 	mockSearch := new(MockSearchClient)
@@ -159,9 +159,6 @@ func TestIndexer_Start_ConsumeFlow(t *testing.T) {
 
 func TestIndexer_Consume_Delete(t *testing.T) {
 	// Setup similar to above but for DELETE event
-	scheme := runtime.NewScheme()
-	v1alpha1.AddToScheme(scheme)
-
 	policy := &v1alpha1.ResourceIndexPolicy{
 		ObjectMeta: metav1.ObjectMeta{Name: "pod-policy"},
 		Spec: v1alpha1.ResourceIndexPolicySpec{
@@ -173,9 +170,12 @@ func TestIndexer_Consume_Delete(t *testing.T) {
 		},
 	}
 
-	k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(policy).Build()
-	policyCache, _ := NewPolicyCache(k8sClient, 1*time.Minute)
-	policyCache.refresh(context.Background())
+	env2, _ := internalcel.NewEnv()
+	policyCache := &PolicyCache{
+		policies: make(map[string]*policyevaluation.CachedPolicy),
+		celEnv:   env2,
+	}
+	policyCache.upsertPolicy(policy)
 
 	mockSearch := new(MockSearchClient)
 	batcher := NewBatcher(mockSearch, BatchConfig{BatchSize: 1, FlushInterval: 1 * time.Minute})