feat: introduce SearchQuery agains meilisearch

Author: JoseSzycho

Taskfile.yaml

@@ -497,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 \
@@ -506,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

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 {

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/overlays/ci/kustomization.yaml

@@ -62,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/dev/kustomization.yaml

@@ -52,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/samples/search_v1alpha1_searchquery.yaml

@@ -0,0 +1,9 @@
+apiVersion: search.miloapis.com/v1alpha1
+kind: SearchQuery
+metadata:
+  name: configmap-search-sample
+spec:
+
+  query: "backend-api" # Empty query returns everything, or put a specific string
+  limit: 100
+  offset: 0

internal/apiserver/apiserver.go

@@ -2,6 +2,8 @@ package apiserver
 
 import (
 	"context"
+	"fmt"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -11,11 +13,13 @@ import (
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/klog/v2"
 
+	"go.miloapis.net/search/internal/indexer"
 	_ "go.miloapis.net/search/internal/metrics"
 	"go.miloapis.net/search/internal/registry/policy/resourceindexpolicy"
+	"go.miloapis.net/search/internal/registry/searchquery"
 	"go.miloapis.net/search/pkg/apis/search/install"
-	searchinstall "go.miloapis.net/search/pkg/apis/search/install"
 	searchv1alpha1 "go.miloapis.net/search/pkg/apis/search/v1alpha1"
+	"go.miloapis.net/search/pkg/meilisearch"
 )
 
 var (
@@ -27,7 +31,6 @@ var (
 
 func init() {
 	install.Install(Scheme)
-	searchinstall.Install(Scheme)
 
 	metav1.AddToGroupVersion(Scheme, schema.GroupVersion{Version: "v1"})
 
@@ -36,6 +39,8 @@ func init() {
 	Scheme.AddKnownTypes(schema.GroupVersion{Group: searchv1alpha1.GroupName, Version: runtime.APIVersionInternal},
 		&searchv1alpha1.ResourceIndexPolicy{},
 		&searchv1alpha1.ResourceIndexPolicyList{},
+		&searchv1alpha1.SearchQuery{},
+		&searchv1alpha1.SearchQueryList{},
 	)
 
 	// Register unversioned meta types required by the API machinery.
@@ -51,7 +56,12 @@ func init() {
 
 // ExtraConfig extends the generic apiserver configuration with search-specific settings.
 type ExtraConfig struct {
-	// Add custom configuration here as needed
+	MeiliClient        *meilisearch.SDKClient
+	PolicyCache        *indexer.PolicyCache
+	MaxSearchLimit     int
+	DefaultSearchLimit int
+	PagingSecret       []byte
+	PagingTimeout      time.Duration
 }
 
 // Config combines generic and search-specific configuration.
@@ -110,12 +120,42 @@ func (c completedConfig) New() (*SearchServer, error) {
 	searchV1alpha1Storage["resourceindexpolicies"] = policyStorage.Store
 	searchV1alpha1Storage["resourceindexpolicies/status"] = policyStorage.Status
 
+	// Add searchquery resources
+	searchqueryStorage := searchquery.NewREST(
+		c.ExtraConfig.MeiliClient,
+		c.ExtraConfig.PolicyCache,
+		c.ExtraConfig.MaxSearchLimit,
+		c.ExtraConfig.DefaultSearchLimit,
+		c.ExtraConfig.PagingSecret,
+		c.ExtraConfig.PagingTimeout,
+	)
+	searchV1alpha1Storage["searchqueries"] = searchqueryStorage
+
 	searchAPIGroupInfo.VersionedResourcesStorageMap["v1alpha1"] = searchV1alpha1Storage
 
 	if err := s.GenericAPIServer.InstallAPIGroup(&searchAPIGroupInfo); err != nil {
 		return nil, err
 	}
 
+	// Add PostStartHook to start policy cache
+	s.GenericAPIServer.AddPostStartHookOrDie("search-policy-cache", func(ctx genericapiserver.PostStartHookContext) error {
+		if err := c.ExtraConfig.PolicyCache.RegisterHandlers(ctx.Context); err != nil {
+			return fmt.Errorf("failed to register policy cache handlers: %w", err)
+		}
+
+		go func() {
+			klog.Info("Starting Search policy cache...")
+			if err := c.ExtraConfig.PolicyCache.Start(ctx.Context); err != nil {
+				klog.Errorf("Policy cache stopped with error: %v", err)
+			}
+		}()
+
+		if !c.ExtraConfig.PolicyCache.WaitForCacheSync(ctx.Context) {
+			return fmt.Errorf("failed to wait for policy cache sync")
+		}
+		return nil
+	})
+
 	klog.Info("Search server initialized successfully")
 
 	return s, nil

internal/indexer/policy_cache.go

@@ -155,6 +155,16 @@ func (c *PolicyCache) deletePolicy(name string) {
 	klog.Infof("Policy %s removed from cache", name)
 }
 
+// Start starts the underlying cache.
+func (c *PolicyCache) Start(ctx context.Context) error {
+	return c.cache.Start(ctx)
+}
+
+// WaitForCacheSync waits for the underlying cache to sync.
+func (c *PolicyCache) WaitForCacheSync(ctx context.Context) bool {
+	return c.cache.WaitForCacheSync(ctx)
+}
+
 // GetPolicies returns a snapshot of all cached policies.
 func (c *PolicyCache) GetPolicies() []*policyevaluation.CachedPolicy {
 	c.mu.RLock()