UnifiedSearch: Introduce a ResourceIndex interface and bleve implementation (#96826)

Co-authored-by: Scott Lepper <scott.lepper@gmail.com>

pkg/storage/unified/resource/document.go

@@ -159,11 +159,8 @@ func NewIndexableDocument(key *ResourceKey, rv int64, obj utils.GrafanaMetaAcces
 	return doc
 }
 
-func StandardDocumentBuilder() DocumentBuilderInfo {
-	return DocumentBuilderInfo{
-		Builder: &standardDocumentBuilder{},
-		Fields:  StandardSearchFields(),
-	}
+func StandardDocumentBuilder() DocumentBuilder {
+	return &standardDocumentBuilder{}
 }
 
 type standardDocumentBuilder struct{}
@@ -295,6 +292,30 @@ func StandardSearchFields() SearchableDocumentFields {
 					FreeText: true,
 				},
 			},
+			{
+				Name:        SEARCH_FIELD_TAGS,
+				Type:        ResourceTableColumnDefinition_STRING,
+				IsArray:     true,
+				Description: "Unique tags",
+				Properties: &ResourceTableColumnDefinition_Properties{
+					Filterable: true,
+				},
+			},
+			{
+				Name:        SEARCH_FIELD_FOLDER,
+				Type:        ResourceTableColumnDefinition_STRING,
+				Description: "Kubernetes name for the folder",
+			},
+			{
+				Name:        SEARCH_FIELD_RV,
+				Type:        ResourceTableColumnDefinition_INT64,
+				Description: "resource version",
+			},
+			{
+				Name:        SEARCH_FIELD_CREATED,
+				Type:        ResourceTableColumnDefinition_INT64,
+				Description: "created timestamp", // date?
+			},
 		})
 		if err != nil {
 			panic("failed to initialize standard search fields")

pkg/storage/unified/resource/document_test.go

@@ -12,7 +12,7 @@ import (
 
 func TestStandardDocumentBuilder(t *testing.T) {
 	ctx := context.Background()
-	builder := StandardDocumentBuilder().Builder
+	builder := StandardDocumentBuilder()
 
 	body, err := os.ReadFile("testdata/playlist-resource.json")
 	require.NoError(t, err)

pkg/storage/unified/resource/resource.proto

@@ -323,6 +323,7 @@ message WatchEvent {
   Resource previous = 4;
 }
 
+// This will soon be deprecated/replaced with ResourceSearchRequest
 message SearchRequest {
   // query string for chosen implementation (currently just bleve)
   string query = 1;
@@ -342,6 +343,92 @@ message SearchRequest {
   repeated string filters = 10;
 }
 
+// Search within a single resource
+message ResourceSearchRequest {
+  message Sort {
+    string field = 1;
+    bool desc = 2; // defaults to ascending
+  }
+  
+  message Facet {
+    string field = 1;
+    int64 limit = 2;
+    // For now, only term queries, eventually?
+    // numeric queries
+    // date queries
+  }
+
+  // The key must include namespace + group + resource
+  ListOptions options = 1;
+
+  // To search additional resource types, add additional keys to this list
+  // NOTE: queries will only support federation across kinds with common fields
+  repeated ResourceKey federated = 2; 
+  
+  // When a query exists, it is parsed and used to influence
+  // query string for chosen implementation (currently just bleve)
+  // The score is only relevant when a query exists
+  string query = 3;
+
+  // max results
+  int64 limit = 4;
+
+  // where to start the query (eg, From)
+  int64 offset = 5;
+
+  // sorting
+  repeated Sort sortBy = 6;
+
+  // calculate field statistics
+  map<string,Facet> facet = 7;
+
+  // the return fields (empty will return everything)
+  repeated string fields = 8;
+
+  // explain each result (added to the each row)
+  bool explain = 9;
+}
+
+message ResourceSearchResponse {
+  message Facet {
+    string field = 1;
+    // The distinct terms
+    int64 total = 2;
+    // The number of documents that do *not* have this field
+    int64 missing = 3;
+    // Top term stats
+    repeated TermFacet terms = 4;
+    // numeric range
+    // date range facets 
+  }
+
+  message TermFacet {
+    string term = 1;
+    int64 count = 2;
+  }
+
+  // Error details
+  ErrorResult error = 1;
+
+  // All results exist within this key
+  ResourceKey key = 2;
+
+  // Query results
+  ResourceTable results = 3;
+
+  // The total hit count
+  uint64 total_hits = 4;
+
+  // indicates how expensive was the query with respect to bytes read
+  uint64 query_cost = 5;
+
+  // maximum score across all fields
+  double max_score = 6;
+
+  // Facet results
+  map<string,Facet> facet = 7;
+}
+
 message GroupBy {
   string name = 1;
   int64 limit = 2;
@@ -352,6 +439,7 @@ message Group {
   int64 count = 2;
 }
 
+// This will soon be deprecated/replaced with ResourceSearchResponse
 message SearchResponse {
   repeated ResourceWrapper items = 1;
   repeated Group groups = 2;

pkg/storage/unified/resource/search.go

@@ -12,6 +12,8 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/grafana/authlib/authz"
 )
 
 type NamespacedResource struct {
@@ -20,8 +22,52 @@ type NamespacedResource struct {
 	Resource  string
 }
 
+// All fields are set
+func (s *NamespacedResource) Valid() bool {
+	return s.Namespace != "" && s.Group != "" && s.Resource != ""
+}
+
+type ResourceIndex interface {
+	// Add a document to the index.  Note it may not be searchable until after flush is called
+	Write(doc *IndexableDocument) error
+
+	// Mark a resource as deleted.  Note it may not be searchable until after flush is called
+	Delete(key *ResourceKey) error
+
+	// Make sure any changes to the index are flushed and available in the next search/origin calls
+	Flush() error
+
+	// Search within a namespaced resource
+	// When working with federated queries, the additional indexes will be passed in explicitly
+	Search(ctx context.Context, access authz.AccessClient, req *ResourceSearchRequest, federate []ResourceIndex) (*ResourceSearchResponse, error)
+
+	// Execute an origin query -- access control is not not checked for each item
+	// NOTE: this will likely be used for provisioning, or it will be removed
+	Origin(ctx context.Context, req *OriginRequest) (*OriginResponse, error)
+}
+
+// SearchBackend contains the technology specific logic to support search
 type SearchBackend interface {
-	// TODO
+	// This will return nil if the key does not exist
+	GetIndex(ctx context.Context, key NamespacedResource) (ResourceIndex, error)
+
+	// Build an index from scratch
+	BuildIndex(ctx context.Context,
+		key NamespacedResource,
+
+		// When the size is known, it will be passed along here
+		// Depending on the size, the backend may choose different options (eg: memory vs disk)
+		size int64,
+
+		// The last known resource version (can be used to know that nothing has changed)
+		resourceVersion int64,
+
+		// The non-standard index fields
+		fields SearchableDocumentFields,
+
+		// The builder will write all documents before returning
+		builder func(index ResourceIndex) (int64, error),
+	) (ResourceIndex, error)
 }
 
 const tracingPrexfixSearch = "unified_search."
@@ -119,7 +165,7 @@ func (s *searchSupport) init(ctx context.Context) error {
 	return nil
 }
 
-func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size int64, rv int64) (any, int64, error) {
+func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size int64, rv int64) (ResourceIndex, int64, error) {
 	_, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Build")
 	defer span.End()
 
@@ -127,10 +173,57 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size
 	if err != nil {
 		return nil, 0, err
 	}
+	fields := s.builders.GetFields(nsr)
 
 	s.log.Debug(fmt.Sprintf("TODO, build %+v (size:%d, rv:%d) // builder:%+v\n", nsr, size, rv, builder))
 
-	return nil, 0, nil
+	key := &ResourceKey{
+		Group:     nsr.Group,
+		Resource:  nsr.Resource,
+		Namespace: nsr.Namespace,
+	}
+	index, err := s.search.BuildIndex(ctx, nsr, size, rv, fields, func(index ResourceIndex) (int64, error) {
+		rv, err = s.storage.ListIterator(ctx, &ListRequest{
+			Limit: 1000000000000, // big number
+			Options: &ListOptions{
+				Key: key,
+			},
+		}, func(iter ListIterator) error {
+			for iter.Next() {
+				if err = iter.Error(); err != nil {
+					return err
+				}
+
+				// Update the key name
+				// Or should we read it from the body?
+				key.Name = iter.Name()
+
+				// Convert it to an indexable document
+				doc, err := builder.BuildDocument(ctx, key, iter.ResourceVersion(), iter.Value())
+				if err != nil {
+					return err
+				}
+
+				// And finally write it to the index
+				if err = index.Write(doc); err != nil {
+					return err
+				}
+			}
+			return err
+		})
+		return rv, err
+	})
+
+	if err != nil {
+		return nil, 0, err
+	}
+
+	if err == nil {
+		err = index.Flush()
+	}
+
+	// rv is the last RV we read.  when watching, we must add all events since that time
+	return index, rv, err
 }
 
 type builderCache struct {
@@ -140,6 +233,9 @@ type builderCache struct {
 	// Possible blob support
 	blob BlobSupport
 
+	// searchable fields initialized once on startup
+	fields map[schema.GroupResource]SearchableDocumentFields
+
 	// lookup by group, then resource (namespace)
 	// This is only modified at startup, so we do not need mutex for access
 	lookup map[string]map[string]DocumentBuilderInfo
@@ -151,6 +247,7 @@ type builderCache struct {
 
 func newBuilderCache(cfg []DocumentBuilderInfo, nsCacheSize int, ttl time.Duration) (*builderCache, error) {
 	cache := &builderCache{
+		fields: make(map[schema.GroupResource]SearchableDocumentFields),
 		lookup: make(map[string]map[string]DocumentBuilderInfo),
 		ns:     expirable.NewLRU[NamespacedResource, DocumentBuilder](nsCacheSize, nil, ttl),
 	}
@@ -173,10 +270,17 @@ func newBuilderCache(cfg []DocumentBuilderInfo, nsCacheSize int, ttl time.Durati
 			cache.lookup[b.GroupResource.Group] = g
 		}
 		g[b.GroupResource.Resource] = b
+
+		// Any custom fields
+		cache.fields[b.GroupResource] = b.Fields
 	}
 	return cache, nil
 }
 
+func (s *builderCache) GetFields(key NamespacedResource) SearchableDocumentFields {
+	return s.fields[schema.GroupResource{Group: key.Group, Resource: key.Resource}]
+}
+
 // context is typically background.  Holds an LRU cache for a
 func (s *builderCache) get(ctx context.Context, key NamespacedResource) (DocumentBuilder, error) {
 	g, ok := s.lookup[key.Group]

pkg/storage/unified/resource/table.go

@@ -7,13 +7,18 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"
 	reflect "reflect"
 	"strconv"
+	"testing"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
 	"github.com/grafana/grafana-plugin-sdk-go/data/utils/jsoniter"
 )
 
@@ -74,6 +79,10 @@ func (x *ResourceTable) ToK8s() (metav1.Table, error) {
 			}
 		} else if r.Key != nil {
 			obj := &metav1.PartialObjectMetadata{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       r.Key.Resource, // :(
+					APIVersion: r.Key.Group,    // :(
+				},
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      r.Key.Name,
 					Namespace: r.Key.Namespace,
@@ -102,6 +111,8 @@ type TableBuilder struct {
 	hasDuplicateNames bool
 }
 
+type ResourceColumnEncoder = func(v any) ([]byte, error)
+
 func NewTableBuilder(cols []*ResourceTableColumnDefinition) (*TableBuilder, error) {
 	table := &TableBuilder{
 		ResourceTable: ResourceTable{
@@ -124,6 +135,15 @@ func NewTableBuilder(cols []*ResourceTableColumnDefinition) (*TableBuilder, erro
 	return table, err
 }
 
+func (x *TableBuilder) Encoders() []ResourceColumnEncoder {
+	encoders := make([]ResourceColumnEncoder, len(x.Columns))
+	for i, f := range x.Columns {
+		v := x.lookup[f.Name]
+		encoders[i] = v.Encode
+	}
+	return encoders
+}
+
 func (x *TableBuilder) AddRow(key *ResourceKey, rv int64, vals map[string]any) error {
 	row := &ResourceTableRow{
 		Key:             key,
@@ -395,6 +415,8 @@ func (x *resourceTableColumn) Encode(v any) ([]byte, error) {
 						f = int64(typed)
 					case float32:
 						f = int64(typed)
+					case float64:
+						f = int64(typed)
 					case uint64:
 						f = int64(typed)
 					case uint:
@@ -547,3 +569,29 @@ func (x *resourceTableColumn) Decode(buff []byte) (any, error) {
 	}
 	return v, err
 }
+
+// AssertTableSnapshot will match a ResourceTable vs the saved value
+func AssertTableSnapshot(t *testing.T, path string, table *ResourceTable) {
+	t.Helper()
+
+	k8sTable, err := table.ToK8s()
+	require.NoError(t, err, "unable to create table response", path)
+	actual, err := json.MarshalIndent(k8sTable, "", "  ")
+	require.NoError(t, err, "unable to write table json", path)
+
+	// Safe to disable, this is a test.
+	// nolint:gosec
+	expected, err := os.ReadFile(path)
+	if err != nil || len(expected) < 1 {
+		assert.Fail(t, "missing file: %s", path)
+	} else if assert.JSONEq(t, string(expected), string(actual)) {
+		return // everything is OK
+	}
+
+	// Write the snapshot
+	// Safe to disable, this is a test.
+	// nolint:gosec
+	err = os.WriteFile(path, actual, 0600)
+	require.NoError(t, err)
+	fmt.Printf("Updated table snapshot: %s\n", path)
+}

pkg/storage/unified/resource/table_test.go

@@ -1,44 +1,15 @@
 package resource
 
 import (
-	"encoding/json"
 	"fmt"
-	"os"
 	"path/filepath"
 	"strings"
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-// AssertTableSnapshot will match a ResourceTable vs the saved value
-func AssertTableSnapshot(t *testing.T, path string, table *ResourceTable) {
-	t.Helper()
-
-	k8sTable, err := table.ToK8s()
-	require.NoError(t, err, "unable to create table response", path)
-	actual, err := json.MarshalIndent(k8sTable, "", "  ")
-	require.NoError(t, err, "unable to write table json", path)
-
-	// Safe to disable, this is a test.
-	// nolint:gosec
-	expected, err := os.ReadFile(path)
-	if err != nil || len(expected) < 1 {
-		assert.Fail(t, "missing file")
-	} else if assert.JSONEq(t, string(expected), string(actual)) {
-		return // everything is OK
-	}
-
-	// Write the snapshot
-	// Safe to disable, this is a test.
-	// nolint:gosec
-	err = os.WriteFile(path, actual, 0600)
-	require.NoError(t, err)
-	fmt.Printf("Updated table snapshot: %s\n", path)
-}
-
 func TestTableFormat(t *testing.T) {
 	columns := []*ResourceTableColumnDefinition{
 		{

pkg/storage/unified/resource/testdata/simple-table.json

@@ -41,6 +41,8 @@
         ]
       ],
       "object": {
+        "kind": "xyz",
+        "apiVersion": "ggg",
         "metadata": {
           "name": "aaa",
           "namespace": "default",
@@ -60,6 +62,8 @@
         ]
       ],
       "object": {
+        "kind": "xyz",
+        "apiVersion": "ggg",
         "metadata": {
           "name": "bbb",
           "namespace": "default",