feat: Introduce SearchQuery API and Refine Indexing Infrastructure

Taskfile.yaml

@@ -122,6 +122,7 @@ tasks:
       # Generate RBAC rules for the controllers.
       - echo "Generating RBAC rules for the controllers..."
       - "\"{{.TOOL_DIR}}/controller-gen\" rbac:roleName=milo-controller-manager paths=\"./internal/controllers/...\" output:dir=\"./config/overlays/controller-manager/core-control-plane/rbac\""
+      - "\"{{.TOOL_DIR}}/controller-gen\" rbac:roleName=milo-resource-indexer paths=\"./internal/indexer/...\" output:dir=\"./config/overlays/resource-indexer/core-control-plane/rbac\""
       - task: generate:openapi
     silent: true
 
@@ -376,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
@@ -409,7 +411,9 @@ tasks:
           --metrics-bind-address=:8085 \
           --health-probe-bind-address=:8086 \
           --leader-elect=false \
-          --meilisearch-domain="http://127.0.0.1:7700"
+          --meilisearch-domain="http://127.0.0.1:7700" \
+          --nats-url="nats://127.0.0.1:4222"
+
     silent: true
 
   dev:run-indexer:
@@ -422,21 +426,10 @@ tasks:
         # Trap to kill background processes (port-forward) on exit
         trap 'kill $(jobs -p)' EXIT
 
-        echo "🚀 Starting NATS port-forward in background..."
-        task dev:pf-nats > /dev/null 2>&1 &
-
-        echo "🚀 Starting Meilisearch port-forward in background..."
-        task dev:pf-meilisearch > /dev/null 2>&1 &
-
-        echo "Waiting for port-forward to be ready..."
-        sleep 5
-
         echo "🚀 Running indexer against local NATS..."
         MEILISEARCH_API_KEY=search-master-key \
         go run ./cmd/search indexer \
           --nats-url="nats://127.0.0.1:4222" \
-          --nats-subject="audit.>" \
-          --nats-queue-group="search-indexer" \
           --nats-audit-consumer-name="search-indexer" \
           --nats-reindex-consumer-name="search-reindexer" \
           --nats-stream-name="AUDIT_EVENTS" \
@@ -472,6 +465,15 @@ tasks:
       - echo "Port forwarding Etcd to localhost:2379..."
       - task test-infra:kubectl -- port-forward -n etcd-system svc/etcd 2379:2379
 
+  dev:pf-all:
+    desc: Port forward all dependencies for local development
+    cmds:
+      - |
+        task dev:pf-meilisearch &
+        task dev:pf-nats &
+        task dev:pf-etcd &
+        wait
+
   dev:run-apiserver:
     desc: Run the API server locally (requires dev:pf-etcd running)
     cmds:
@@ -495,6 +497,7 @@ tasks:
       - |
         # Use KUBECONFIG if set, otherwise fallback to default
         KCFG=${KUBECONFIG:-$HOME/.kube/config}
+        export MEILISEARCH_API_KEY=search-master-key
         go run ./cmd/search serve \
           --etcd-servers http://127.0.0.1:2379 \
           --secure-port 9443 \
@@ -504,7 +507,9 @@ tasks:
           --authorization-kubeconfig="$KCFG" \
           --kubeconfig="$KCFG" \
           --client-ca-file="{{.CERTS_DIR}}/kind-ca.crt" \
-          --authorization-always-allow-paths=/healthz,/readyz,/livez,/openapi,/openapi/v2,/openapi/v3,/apis,/api
+          --authorization-always-allow-paths=/healthz,/readyz,/livez,/openapi,/openapi/v2,/openapi/v3,/apis,/api \
+          --meilisearch-domain="http://127.0.0.1:7700" \
+
 
   dev:undeploy:
     desc: Undeploy Search server from test-infra cluster
@@ -660,7 +665,6 @@ tasks:
     deps:
       - dev:build
       - dev:load
-      - dev:deploy
     cmds:
       - |
         set -e
@@ -671,15 +675,19 @@ 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
     cmds:
-      - task test-infra:kubectl -- exec -n nats-system $(task test-infra:kubectl -- get pods -n nats-system -l app.kubernetes.io/component=nats-box -o jsonpath="{.items[0].metadata.name}") -- nats consumer info AUDIT_EVENTS search-indexer
+      - task test-infra:kubectl -- exec -n nats-system $(task test-infra:kubectl -- get pods -n nats-system -l app.kubernetes.io/component=nats-box -o jsonpath="{.items[0].metadata.name}") -- nats consumer info AUDIT_EVENTS search-indexer
+      - task test-infra:kubectl -- exec -n nats-system $(task test-infra:kubectl -- get pods -n nats-system -l app.kubernetes.io/component=nats-box -o jsonpath="{.items[0].metadata.name}") -- nats consumer info REINDEX_EVENTS search-reindexer

cmd/search/indexer/command.go

@@ -17,6 +17,7 @@ import (
 	"k8s.io/klog/v2"
 	runtimecache "sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 // ResourceIndexerOptions holds the configuration for the resource indexer.
@@ -159,6 +160,7 @@ func NewIndexerCommand() *cobra.Command {
 
 // Run starts the indexer consumer
 func Run(o *ResourceIndexerOptions, ctx context.Context) error {
+	ctrllog.SetLogger(klog.NewKlogr())
 	// Build a scheme and REST config for the controller-runtime cache.
 	scheme := runtime.NewScheme()
 	if err := searchv1alpha1.AddToScheme(scheme); err != nil {
@@ -188,18 +190,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)

cmd/search/main.go

@@ -5,12 +5,15 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"time"
 
+	"github.com/google/uuid"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	searchapiserver "go.miloapis.net/search/internal/apiserver"
 	"go.miloapis.net/search/internal/version"
 	searchv1alpha1 "go.miloapis.net/search/pkg/apis/search/v1alpha1"
+	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	apiopenapi "k8s.io/apiserver/pkg/endpoints/openapi"
 	genericapiserver "k8s.io/apiserver/pkg/server"
@@ -22,9 +25,13 @@ import (
 	logsapi "k8s.io/component-base/logs/api/v1"
 	"k8s.io/klog/v2"
 	"k8s.io/kube-openapi/pkg/common"
+	runtimecache "sigs.k8s.io/controller-runtime/pkg/cache"
+	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
 
 	"go.miloapis.net/search/cmd/search/indexer"
 	"go.miloapis.net/search/cmd/search/manager"
+	internalindexer "go.miloapis.net/search/internal/indexer"
+	"go.miloapis.net/search/pkg/meilisearch"
 
 	// Register JSON logging format
 	_ "k8s.io/component-base/logs/json/register"
@@ -140,8 +147,14 @@ func NewVersionCommand() *cobra.Command {
 
 // SearchServerOptions contains configuration for the search server.
 type SearchServerOptions struct {
-	RecommendedOptions *options.RecommendedOptions
-	Logs               *logsapi.LoggingConfiguration
+	RecommendedOptions         *options.RecommendedOptions
+	Logs                       *logsapi.LoggingConfiguration
+	MeilisearchDomain          string
+	MeilisearchHTTPTimeout     time.Duration
+	MeilisearchTaskWaitTimeout time.Duration
+	MaxSearchLimit             int
+	DefaultSearchLimit         int
+	PagingTimeout              time.Duration
 }
 
 // NewSearchServerOptions creates options with default values.
@@ -151,14 +164,23 @@ func NewSearchServerOptions() *SearchServerOptions {
 			"/registry/search.miloapis.com",
 			searchapiserver.Codecs.LegacyCodec(searchapiserver.Scheme.PrioritizedVersionsAllGroups()...),
 		),
-		Logs: logsapi.NewLoggingConfiguration(),
+		Logs:                   logsapi.NewLoggingConfiguration(),
+		MeilisearchDomain:      "http://meilisearch.meilisearch-system.svc.cluster.local:7700",
+		MeilisearchHTTPTimeout: 60 * time.Second,
+		MaxSearchLimit:         100,
+		DefaultSearchLimit:     10,
+		PagingTimeout:          24 * time.Hour,
 	}
 
 	return o
 }
 
 func (o *SearchServerOptions) AddFlags(fs *pflag.FlagSet) {
 	o.RecommendedOptions.AddFlags(fs)
+	fs.StringVar(&o.MeilisearchDomain, "meilisearch-domain", o.MeilisearchDomain, "Domain of the Meilisearch instance.")
+	fs.IntVar(&o.MaxSearchLimit, "max-search-limit", o.MaxSearchLimit, "The maximum number of results a SearchQuery can return in a single request.")
+	fs.IntVar(&o.DefaultSearchLimit, "default-search-limit", o.DefaultSearchLimit, "The default number of results a SearchQuery returns when no limit is specified.")
+	fs.DurationVar(&o.PagingTimeout, "paging-timeout", o.PagingTimeout, "The duration for which a paging (continue) token is valid.")
 }
 
 func (o *SearchServerOptions) Complete() error {
@@ -197,9 +219,53 @@ func (o *SearchServerOptions) Config() (*searchapiserver.Config, error) {
 		return nil, fmt.Errorf("failed to apply recommended options: %w", err)
 	}
 
+	meiliClient, err := meilisearch.NewSDKClient(meilisearch.SDKConfig{
+		Domain:      o.MeilisearchDomain,
+		APIKey:      os.Getenv("MEILISEARCH_API_KEY"),
+		WaitTimeout: o.MeilisearchTaskWaitTimeout,
+		HTTPTimeout: o.MeilisearchHTTPTimeout,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize meilisearch client: %w", err)
+	}
+
+	// Create a dedicated scheme for the policy cache that only contains versioned types.
+	// This avoids ambiguity when the main scheme registers the same type for both
+	// v1alpha1 and __internal versions.
+	policyScheme := runtime.NewScheme()
+	if err := searchv1alpha1.AddToScheme(policyScheme); err != nil {
+		return nil, fmt.Errorf("failed to add v1alpha1 to policy scheme: %w", err)
+	}
+
+	// Create a controller-runtime cache that uses a watch stream (informer)
+	// to keep ResourceIndexPolicies in-sync.
+	k8sCache, err := runtimecache.New(genericConfig.LoopbackClientConfig, runtimecache.Options{Scheme: policyScheme})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create controller-runtime cache: %w", err)
+	}
+
+	// Create the policy cache (strict ready checking enabled)
+	indexPolicyCache, err := internalindexer.NewPolicyCache(k8sCache, true)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create policy cache: %w", err)
+	}
+
+	pagingSecret := []byte(os.Getenv("SEARCH_PAGING_SECRET"))
+	if len(pagingSecret) == 0 {
+		klog.Info("SEARCH_PAGING_SECRET not set, generating a random one")
+		pagingSecret = []byte(uuid.New().String())
+	}
+
 	serverConfig := &searchapiserver.Config{
 		GenericConfig: genericConfig,
-		ExtraConfig:   searchapiserver.ExtraConfig{},
+		ExtraConfig: searchapiserver.ExtraConfig{
+			MeiliClient:        meiliClient,
+			PolicyCache:        indexPolicyCache,
+			MaxSearchLimit:     o.MaxSearchLimit,
+			DefaultSearchLimit: o.DefaultSearchLimit,
+			PagingSecret:       pagingSecret,
+			PagingTimeout:      o.PagingTimeout,
+		},
 	}
 
 	return serverConfig, nil
@@ -211,6 +277,8 @@ func Run(options *SearchServerOptions, ctx context.Context) error {
 		return fmt.Errorf("failed to apply logging configuration: %w", err)
 	}
 
+	ctrllog.SetLogger(klog.NewKlogr())
+
 	config, err := options.Config()
 	if err != nil {
 		return err

config/base/apiserver/deployment.yaml

@@ -97,6 +97,9 @@ spec:
           value: "4"
         - name: TRACING_CONFIG_FILE
           value: ""
+        envFrom:
+        - secretRef:
+            name: search-apiserver
         # TEMPLATE NOTE: Add your service-specific environment variables here
         # Example: database credentials, API keys, configuration values
         resources:

config/base/resource-indexer/deployment.yaml

@@ -2,7 +2,6 @@ apiVersion: apps/v1
 kind: Deployment
 metadata:
   name: resource-indexer
-  namespace: search-system
   labels:
     app.kubernetes.io/name: resource-indexer
     app.kubernetes.io/component: indexer

config/overlays/ci/kustomization.yaml

@@ -5,9 +5,10 @@ namespace: search-system
 
 resources:
   - ../../base/apiserver
-  - ../../base/resource-indexer
   # Controller manager with RBAC
   - ../controller-manager/core-control-plane
+  # Resource indexer with RBAC
+  - ../resource-indexer/core-control-plane
 
 # Include components for CI environment
 components:
@@ -61,3 +62,6 @@ secretGenerator:
 - name: search-indexer
   literals:
   - MEILISEARCH_API_KEY=search-master-key
+- name: search-apiserver
+  literals:
+  - MEILISEARCH_API_KEY=search-master-key

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

@@ -4,6 +4,14 @@ kind: ClusterRole
 metadata:
   name: milo-controller-manager
 rules:
+- apiGroups:
+  - '*'
+  resources:
+  - '*'
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - search.miloapis.com
   resources:

config/overlays/dev/kustomization.yaml

@@ -5,9 +5,10 @@ namespace: search-system
 
 resources:
   - ../../base/apiserver
-  - ../../base/resource-indexer
   # Controller manager with RBAC
   - ../controller-manager/core-control-plane
+  # Resource indexer with RBAC
+  - ../resource-indexer/core-control-plane
 
 # Include components for development environment
 components:
@@ -51,4 +52,6 @@ secretGenerator:
 - name: search-indexer
   literals:
   - MEILISEARCH_API_KEY=search-master-key
-
+- name: search-apiserver
+  literals:
+  - MEILISEARCH_API_KEY=search-master-key

config/overlays/resource-indexer/core-control-plane/kustomization.yaml

@@ -0,0 +1,14 @@
+# This overlay is used to install the resource indexer into the core control
+# plane.
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  # Install the base resource indexer resources.
+  - ../../../base/resource-indexer
+  # Install the RBAC resources for the resource indexer that are specific to
+  # the core control plane.
+  - rbac
+
+patches:
+- path: patches/deployment.yaml