UnifiedSearch: Introduce a DocumentBuilder interface (#96738)
This commit is contained in:
Ryan McKinley
@@ -0,0 +1,320 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
)
|
||||
|
||||
// Convert raw resource bytes into an IndexableDocument
|
||||
type DocumentBuilder interface {
|
||||
// Convert raw bytes into an document that can be written
|
||||
BuildDocument(ctx context.Context, key *ResourceKey, rv int64, value []byte) (*IndexableDocument, error)
|
||||
}
|
||||
|
||||
// Registry of the searchable document fields
|
||||
type SearchableDocumentFields interface {
|
||||
Fields() []string
|
||||
Field(name string) *ResourceTableColumnDefinition
|
||||
}
|
||||
|
||||
// Some kinds will require special processing for their namespace
|
||||
type NamespacedDocumentSupplier = func(ctx context.Context, namespace string, blob BlobSupport) (DocumentBuilder, error)
|
||||
|
||||
// Register how documents can be built for a resource
|
||||
type DocumentBuilderInfo struct {
|
||||
// The target resource (empty will be used to match anything)
|
||||
GroupResource schema.GroupResource
|
||||
|
||||
// Defines the searchable fields
|
||||
// NOTE: this does not include the root/common fields, only values specific to the the builder
|
||||
Fields SearchableDocumentFields
|
||||
|
||||
// simple/static builders that do not depend on the environment can be declared once
|
||||
Builder DocumentBuilder
|
||||
|
||||
// Complicated builders (eg dashboards!) will be declared dynamically and managed by the ResourceServer
|
||||
Namespaced NamespacedDocumentSupplier
|
||||
}
|
||||
|
||||
type DocumentBuilderSupplier interface {
|
||||
GetDocumentBuilders() ([]DocumentBuilderInfo, error)
|
||||
}
|
||||
|
||||
// IndexableDocument can be written to a ResourceIndex
|
||||
// Although public, this is *NOT* an end user interface
|
||||
type IndexableDocument struct {
|
||||
// The resource key
|
||||
Key *ResourceKey `json:"key"`
|
||||
|
||||
// Resource version for the resource (if known)
|
||||
RV int64 `json:"rv,omitempty"`
|
||||
|
||||
// The generic display name
|
||||
Title string `json:"title,omitempty"`
|
||||
|
||||
// A generic description -- helpful in global search
|
||||
Description string `json:"description,omitempty"`
|
||||
|
||||
// Like dashboard tags
|
||||
Tags []string `json:"tags,omitempty"`
|
||||
|
||||
// Generic metadata labels
|
||||
Labels map[string]string `json:"labels,omitempty"`
|
||||
|
||||
// The folder (K8s name)
|
||||
Folder string `json:"folder,omitempty"`
|
||||
|
||||
// The first time this resource was saved
|
||||
Created int64 `json:"created,omitempty"`
|
||||
|
||||
// Who created the resource (will be in the form `user:uid`)
|
||||
CreatedBy string `json:"createdBy,omitempty"`
|
||||
|
||||
// The last time a user updated the spec
|
||||
Updated int64 `json:"updated,omitempty"`
|
||||
|
||||
// Who updated the resource (will be in the form `user:uid`)
|
||||
UpdatedBy string `json:"updatedBy,omitempty"`
|
||||
|
||||
// Searchable nested keys
|
||||
// The key should exist from the fields defined in DocumentBuilderInfo
|
||||
// This should not contain duplicate information from the results above
|
||||
// The meaning of these fields changes depending on the field type
|
||||
// These values typically come from the Spec, but may also come from status
|
||||
// metadata, annotations, or external data linked at index time
|
||||
Fields map[string]any `json:"fields,omitempty"`
|
||||
|
||||
// Maintain a list of resource references.
|
||||
// Someday this will likely be part of https://github.com/grafana/gamma
|
||||
References ResourceReferences `json:"reference,omitempty"`
|
||||
|
||||
// When the resource is managed by an upstream repository
|
||||
RepoInfo *utils.ResourceRepositoryInfo `json:"repository,omitempty"`
|
||||
}
|
||||
|
||||
type ResourceReference struct {
|
||||
Relation string `json:"relation"` // eg: depends-on
|
||||
Group string `json:"group,omitempty"` // the api group
|
||||
Version string `json:"version,omitempty"` // the api version
|
||||
Kind string `json:"kind,omitempty"` // panel, data source (for now)
|
||||
Name string `json:"name"` // the UID / panel name
|
||||
}
|
||||
|
||||
func (m ResourceReference) String() string {
|
||||
var sb strings.Builder
|
||||
sb.WriteString(m.Relation)
|
||||
sb.WriteString(">>")
|
||||
sb.WriteString(m.Group)
|
||||
if m.Version != "" {
|
||||
sb.WriteString("/")
|
||||
sb.WriteString(m.Version)
|
||||
}
|
||||
if m.Kind != "" {
|
||||
sb.WriteString("/")
|
||||
sb.WriteString(m.Kind)
|
||||
}
|
||||
sb.WriteString("/")
|
||||
sb.WriteString(m.Name)
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// Sortable list of references
|
||||
type ResourceReferences []ResourceReference
|
||||
|
||||
func (m ResourceReferences) Len() int { return len(m) }
|
||||
func (m ResourceReferences) Swap(i, j int) { m[i], m[j] = m[j], m[i] }
|
||||
func (m ResourceReferences) Less(i, j int) bool {
|
||||
a := m[i].String()
|
||||
b := m[j].String()
|
||||
return strings.Compare(a, b) > 0
|
||||
}
|
||||
|
||||
// Create a new indexable document based on a generic k8s resource
|
||||
func NewIndexableDocument(key *ResourceKey, rv int64, obj utils.GrafanaMetaAccessor) *IndexableDocument {
|
||||
doc := &IndexableDocument{
|
||||
Key: key,
|
||||
RV: rv,
|
||||
Title: obj.FindTitle(key.Name), // We always want *something* to display
|
||||
Labels: obj.GetLabels(),
|
||||
Folder: obj.GetFolder(),
|
||||
CreatedBy: obj.GetCreatedBy(),
|
||||
UpdatedBy: obj.GetUpdatedBy(),
|
||||
}
|
||||
doc.RepoInfo, _ = obj.GetRepositoryInfo()
|
||||
ts := obj.GetCreationTimestamp()
|
||||
if !ts.Time.IsZero() {
|
||||
doc.Created = ts.Time.UnixMilli()
|
||||
}
|
||||
tt, err := obj.GetUpdatedTimestamp()
|
||||
if err != nil && tt != nil {
|
||||
doc.Updated = tt.UnixMilli()
|
||||
}
|
||||
return doc
|
||||
}
|
||||
|
||||
func StandardDocumentBuilder() DocumentBuilderInfo {
|
||||
return DocumentBuilderInfo{
|
||||
Builder: &standardDocumentBuilder{},
|
||||
Fields: StandardSearchFields(),
|
||||
}
|
||||
}
|
||||
|
||||
type standardDocumentBuilder struct{}
|
||||
|
||||
func (s *standardDocumentBuilder) BuildDocument(ctx context.Context, key *ResourceKey, rv int64, value []byte) (*IndexableDocument, error) {
|
||||
tmp := &unstructured.Unstructured{}
|
||||
err := tmp.UnmarshalJSON(value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
obj, err := utils.MetaAccessor(tmp)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
doc := NewIndexableDocument(key, rv, obj)
|
||||
doc.Title = obj.FindTitle(key.Name)
|
||||
return doc, nil
|
||||
}
|
||||
|
||||
type searchableDocumentFields struct {
|
||||
names []string
|
||||
fields map[string]*resourceTableColumn
|
||||
}
|
||||
|
||||
// This requires unique names
|
||||
func NewSearchableDocumentFields(columns []*ResourceTableColumnDefinition) (SearchableDocumentFields, error) {
|
||||
f := &searchableDocumentFields{
|
||||
names: make([]string, len(columns)),
|
||||
fields: make(map[string]*resourceTableColumn),
|
||||
}
|
||||
for i, c := range columns {
|
||||
if f.fields[c.Name] != nil {
|
||||
return nil, fmt.Errorf("duplicate name")
|
||||
}
|
||||
col, err := newResourceTableColumn(c, i)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
f.names[i] = c.Name
|
||||
f.fields[c.Name] = col
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
func (x *searchableDocumentFields) Fields() []string {
|
||||
return x.names
|
||||
}
|
||||
|
||||
func (x *searchableDocumentFields) Field(name string) *ResourceTableColumnDefinition {
|
||||
f, ok := x.fields[name]
|
||||
if ok {
|
||||
return f.def
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
const SEARCH_FIELD_ID = "_id" // {namespace}/{group}/{resource}/{name}
|
||||
const SEARCH_FIELD_GROUP_RESOURCE = "gr" // group/resource
|
||||
const SEARCH_FIELD_NAMESPACE = "namespace"
|
||||
const SEARCH_FIELD_NAME = "name"
|
||||
const SEARCH_FIELD_RV = "rv"
|
||||
const SEARCH_FIELD_TITLE = "title"
|
||||
const SEARCH_FIELD_DESCRIPTION = "description"
|
||||
const SEARCH_FIELD_TAGS = "tags"
|
||||
const SEARCH_FIELD_LABELS = "labels" // All labels, not a specific one
|
||||
|
||||
const SEARCH_FIELD_FOLDER = "folder"
|
||||
const SEARCH_FIELD_CREATED = "created"
|
||||
const SEARCH_FIELD_CREATED_BY = "createdBy"
|
||||
const SEARCH_FIELD_UPDATED = "updated"
|
||||
const SEARCH_FIELD_UPDATED_BY = "updatedBy"
|
||||
const SEARCH_FIELD_REPOSITORY = "repository"
|
||||
const SEARCH_FIELD_REPOSITORY_HASH = "repository_hash"
|
||||
|
||||
const SEARCH_FIELD_SCORE = "_score" // the match score
|
||||
const SEARCH_FIELD_EXPLAIN = "_explain" // score explanation as JSON object
|
||||
|
||||
var standardSearchFieldsInit sync.Once
|
||||
var standardSearchFields SearchableDocumentFields
|
||||
|
||||
func StandardSearchFields() SearchableDocumentFields {
|
||||
standardSearchFieldsInit.Do(func() {
|
||||
var err error
|
||||
standardSearchFields, err = NewSearchableDocumentFields([]*ResourceTableColumnDefinition{
|
||||
{
|
||||
Name: SEARCH_FIELD_ID,
|
||||
Type: ResourceTableColumnDefinition_STRING,
|
||||
Description: "Unique Identifier. {namespace}/{group}/{resource}/{name}",
|
||||
Properties: &ResourceTableColumnDefinition_Properties{
|
||||
NotNull: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: SEARCH_FIELD_GROUP_RESOURCE,
|
||||
Type: ResourceTableColumnDefinition_STRING,
|
||||
Description: "The resource kind: {group}/{resource}",
|
||||
Properties: &ResourceTableColumnDefinition_Properties{
|
||||
NotNull: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: SEARCH_FIELD_NAMESPACE,
|
||||
Type: ResourceTableColumnDefinition_STRING,
|
||||
Description: "Tenant isolation",
|
||||
Properties: &ResourceTableColumnDefinition_Properties{
|
||||
NotNull: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: SEARCH_FIELD_NAME,
|
||||
Type: ResourceTableColumnDefinition_STRING,
|
||||
Description: "Kubernetes name. Unique identifier within a namespace+group+resource",
|
||||
Properties: &ResourceTableColumnDefinition_Properties{
|
||||
NotNull: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: SEARCH_FIELD_TITLE,
|
||||
Type: ResourceTableColumnDefinition_STRING,
|
||||
Description: "Display name for the resource",
|
||||
},
|
||||
{
|
||||
Name: SEARCH_FIELD_DESCRIPTION,
|
||||
Type: ResourceTableColumnDefinition_STRING,
|
||||
Description: "An account of the resource.",
|
||||
Properties: &ResourceTableColumnDefinition_Properties{
|
||||
FreeText: true,
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
panic("failed to initialize standard search fields")
|
||||
}
|
||||
})
|
||||
return standardSearchFields
|
||||
}
|
||||
|
||||
// // Helper function to convert everything except the "Fields" property to values
|
||||
// // NOTE: this is really to help testing things absent real backend index
|
||||
// func IndexableDocumentStandardFields(doc *IndexableDocument) map[string]any {
|
||||
// fields := make(map[string]any)
|
||||
|
||||
// // These should always exist
|
||||
// fields[SEARCH_FIELD_ID] = doc.Key.SearchID()
|
||||
// fields[SEARCH_FIELD_NAMESPACE] = doc.Key.Namespace
|
||||
// fields[SEARCH_FIELD_NAME] = doc.Key.Name
|
||||
// fields[SEARCH_FIELD_GROUP_RESOURCE] = fmt.Sprintf("%s/%s", doc.Key.Group, doc.Key.Resource)
|
||||
|
||||
// fields[SEARCH_FIELD_TITLE] = doc.Title
|
||||
|
||||
// return fields
|
||||
// }
|
||||
@@ -0,0 +1,47 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestStandardDocumentBuilder(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
builder := StandardDocumentBuilder().Builder
|
||||
|
||||
body, err := os.ReadFile("testdata/playlist-resource.json")
|
||||
require.NoError(t, err)
|
||||
doc, err := builder.BuildDocument(ctx, &ResourceKey{
|
||||
Namespace: "default",
|
||||
Group: "playlists.grafana.app",
|
||||
Resource: "playlists",
|
||||
Name: "test1",
|
||||
}, 10, body)
|
||||
require.NoError(t, err)
|
||||
|
||||
jj, _ := json.MarshalIndent(doc, "", " ")
|
||||
fmt.Printf("%s\n", string(jj))
|
||||
require.JSONEq(t, `{
|
||||
"key": {
|
||||
"namespace": "default",
|
||||
"group": "playlists.grafana.app",
|
||||
"resource": "playlists",
|
||||
"name": "test1"
|
||||
},
|
||||
"rv": 10,
|
||||
"title": "test1",
|
||||
"created": 1717236672000,
|
||||
"createdBy": "user:ABC",
|
||||
"updatedBy": "user:XYZ",
|
||||
"repository": {
|
||||
"name": "SQL",
|
||||
"path": "15",
|
||||
"hash": "xyz"
|
||||
}
|
||||
}`, string(jj))
|
||||
}
|
||||
@@ -1,5 +1,10 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func verifyRequestKey(key *ResourceKey) *ErrorResult {
|
||||
if key == nil {
|
||||
return NewBadRequestError("missing resource key")
|
||||
@@ -28,3 +33,39 @@ func matchesQueryKey(query *ResourceKey, key *ResourceKey) bool {
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
const clusterNamespace = "**cluster**"
|
||||
|
||||
// Convert the key to a search ID string
|
||||
func (x *ResourceKey) SearchID() string {
|
||||
var sb strings.Builder
|
||||
if x.Namespace == "" {
|
||||
sb.WriteString(clusterNamespace)
|
||||
} else {
|
||||
sb.WriteString(x.Namespace)
|
||||
}
|
||||
sb.WriteString("/")
|
||||
sb.WriteString(x.Group)
|
||||
sb.WriteString("/")
|
||||
sb.WriteString(x.Resource)
|
||||
sb.WriteString("/")
|
||||
sb.WriteString(x.Name)
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func (x *ResourceKey) ReadSearchID(v string) error {
|
||||
parts := strings.Split(v, "/")
|
||||
if len(parts) != 4 {
|
||||
return fmt.Errorf("invalid search id (expecting 3 slashes)")
|
||||
}
|
||||
|
||||
x.Namespace = parts[0]
|
||||
x.Group = parts[1]
|
||||
x.Resource = parts[2]
|
||||
x.Name = parts[3]
|
||||
|
||||
if x.Namespace == clusterNamespace {
|
||||
x.Namespace = ""
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -19,3 +19,41 @@ func TestKeyMatching(t *testing.T) {
|
||||
}))
|
||||
})
|
||||
}
|
||||
|
||||
func TestSearchIDKeys(t *testing.T) {
|
||||
tests := []struct {
|
||||
input string
|
||||
expected *ResourceKey // nil error
|
||||
}{
|
||||
{input: "a"}, // error
|
||||
{input: "default/group/resource/name",
|
||||
expected: &ResourceKey{
|
||||
Namespace: "default",
|
||||
Group: "group",
|
||||
Resource: "resource",
|
||||
Name: "name",
|
||||
}},
|
||||
{input: "/group/resource/", // missing name
|
||||
expected: &ResourceKey{
|
||||
Namespace: "",
|
||||
Group: "group",
|
||||
Resource: "resource",
|
||||
Name: "",
|
||||
}},
|
||||
{input: "**cluster**/group/resource/aaa", // cluster namespace
|
||||
expected: &ResourceKey{
|
||||
Namespace: "",
|
||||
Group: "group",
|
||||
Resource: "resource",
|
||||
Name: "aaa",
|
||||
}},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
tmp := &ResourceKey{}
|
||||
err := tmp.ReadSearchID(test.input)
|
||||
if err == nil {
|
||||
require.Equal(t, test.expected, tmp, test.input)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/golang-lru/v2/expirable"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
)
|
||||
|
||||
type NamespacedResource struct {
|
||||
Namespace string
|
||||
Group string
|
||||
Resource string
|
||||
}
|
||||
|
||||
type SearchBackend interface {
|
||||
// TODO
|
||||
}
|
||||
|
||||
const tracingPrexfixSearch = "unified_search."
|
||||
|
||||
// This supports indexing+search regardless of implementation
|
||||
type searchSupport struct {
|
||||
tracer trace.Tracer
|
||||
log *slog.Logger
|
||||
storage StorageBackend
|
||||
search SearchBackend
|
||||
builders *builderCache
|
||||
initWorkers int
|
||||
}
|
||||
|
||||
func newSearchSupport(opts SearchOptions, storage StorageBackend, blob BlobSupport, tracer trace.Tracer) (support *searchSupport, err error) {
|
||||
// No backend search support
|
||||
if opts.Backend == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
if opts.WorkerThreads < 1 {
|
||||
opts.WorkerThreads = 1
|
||||
}
|
||||
|
||||
support = &searchSupport{
|
||||
tracer: tracer,
|
||||
storage: storage,
|
||||
search: opts.Backend,
|
||||
log: slog.Default().With("logger", "resource-search"),
|
||||
initWorkers: opts.WorkerThreads,
|
||||
}
|
||||
|
||||
info, err := opts.Resources.GetDocumentBuilders()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
support.builders, err = newBuilderCache(info, 100, time.Minute*2) // TODO? opts
|
||||
if support.builders != nil {
|
||||
support.builders.blob = blob
|
||||
}
|
||||
|
||||
return support, err
|
||||
}
|
||||
|
||||
// init is called during startup. any failure will block startup and continued execution
|
||||
func (s *searchSupport) init(ctx context.Context) error {
|
||||
_, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Init")
|
||||
defer span.End()
|
||||
|
||||
// TODO, replace namespaces with a query that gets top values
|
||||
namespaces, err := s.storage.Namespaces(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Hardcoded for now... should come from the query
|
||||
kinds := []schema.GroupResource{
|
||||
{Group: "dashboard.grafana.app", Resource: "dashboards"},
|
||||
{Group: "playlist.grafana.app", Resource: "playlists"},
|
||||
}
|
||||
|
||||
totalBatchesIndexed := 0
|
||||
group := errgroup.Group{}
|
||||
group.SetLimit(s.initWorkers)
|
||||
|
||||
// Prepare all the (large) indexes
|
||||
// TODO, threading and query real information:
|
||||
// SELECT namespace,"group",resource,COUNT(*),resource_version FROM resource
|
||||
// GROUP BY "group", "resource", "namespace"
|
||||
// ORDER BY resource_version desc;
|
||||
for _, ns := range namespaces {
|
||||
for _, gr := range kinds {
|
||||
group.Go(func() error {
|
||||
s.log.Debug("initializing search index", "namespace", ns, "gr", gr)
|
||||
totalBatchesIndexed++
|
||||
_, _, err = s.build(ctx, NamespacedResource{
|
||||
Group: gr.Group,
|
||||
Resource: gr.Resource,
|
||||
Namespace: ns,
|
||||
}, 10, 0) // TODO, approximate size
|
||||
return err
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
err = group.Wait()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
span.AddEvent("namespaces indexed", trace.WithAttributes(attribute.Int("namespaced_indexed", totalBatchesIndexed)))
|
||||
|
||||
s.log.Debug("TODO, listen to all events")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size int64, rv int64) (any, int64, error) {
|
||||
_, span := s.tracer.Start(ctx, tracingPrexfixSearch+"Build")
|
||||
defer span.End()
|
||||
|
||||
builder, err := s.builders.get(ctx, nsr)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
s.log.Debug(fmt.Sprintf("TODO, build %+v (size:%d, rv:%d) // builder:%+v\n", nsr, size, rv, builder))
|
||||
|
||||
return nil, 0, nil
|
||||
}
|
||||
|
||||
type builderCache struct {
|
||||
// The default builder
|
||||
defaultBuilder DocumentBuilder
|
||||
|
||||
// Possible blob support
|
||||
blob BlobSupport
|
||||
|
||||
// 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
|
||||
|
||||
// For namespaced based resources that require a cache
|
||||
ns *expirable.LRU[NamespacedResource, DocumentBuilder]
|
||||
mu sync.Mutex // only locked for a cache miss
|
||||
}
|
||||
|
||||
func newBuilderCache(cfg []DocumentBuilderInfo, nsCacheSize int, ttl time.Duration) (*builderCache, error) {
|
||||
cache := &builderCache{
|
||||
lookup: make(map[string]map[string]DocumentBuilderInfo),
|
||||
ns: expirable.NewLRU[NamespacedResource, DocumentBuilder](nsCacheSize, nil, ttl),
|
||||
}
|
||||
if len(cfg) == 0 {
|
||||
return cache, fmt.Errorf("no builders configured")
|
||||
}
|
||||
|
||||
for _, b := range cfg {
|
||||
// the default
|
||||
if b.GroupResource.Group == "" && b.GroupResource.Resource == "" {
|
||||
if b.Builder == nil {
|
||||
return cache, fmt.Errorf("default document builder is missing")
|
||||
}
|
||||
cache.defaultBuilder = b.Builder
|
||||
continue
|
||||
}
|
||||
g, ok := cache.lookup[b.GroupResource.Group]
|
||||
if !ok {
|
||||
g = make(map[string]DocumentBuilderInfo)
|
||||
cache.lookup[b.GroupResource.Group] = g
|
||||
}
|
||||
g[b.GroupResource.Resource] = b
|
||||
}
|
||||
return cache, nil
|
||||
}
|
||||
|
||||
// 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]
|
||||
if ok {
|
||||
r, ok := g[key.Resource]
|
||||
if ok {
|
||||
if r.Builder != nil {
|
||||
return r.Builder, nil
|
||||
}
|
||||
|
||||
// The builder needs context
|
||||
builder, ok := s.ns.Get(key)
|
||||
if ok {
|
||||
return builder, nil
|
||||
}
|
||||
{
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
b, err := r.Namespaced(ctx, key.Namespace, s.blob)
|
||||
if err == nil {
|
||||
_ = s.ns.Add(key, b)
|
||||
}
|
||||
return b, err
|
||||
}
|
||||
}
|
||||
}
|
||||
return s.defaultBuilder, nil
|
||||
}
|
||||
@@ -123,6 +123,18 @@ type BlobConfig struct {
|
||||
Backend BlobSupport
|
||||
}
|
||||
|
||||
// Passed as input to the constructor
|
||||
type SearchOptions struct {
|
||||
// The raw index backend (eg, bleve, frames, parquet, etc)
|
||||
Backend SearchBackend
|
||||
|
||||
// The supported resource types
|
||||
Resources DocumentBuilderSupplier
|
||||
|
||||
// How many threads should build indexes
|
||||
WorkerThreads int
|
||||
}
|
||||
|
||||
type ResourceServerOptions struct {
|
||||
// OTel tracer
|
||||
Tracer trace.Tracer
|
||||
@@ -136,6 +148,9 @@ type ResourceServerOptions struct {
|
||||
// Requests based on a search index
|
||||
Index ResourceIndexServer
|
||||
|
||||
// Search options
|
||||
Search SearchOptions
|
||||
|
||||
// Diagnostics
|
||||
Diagnostics DiagnosticsServer
|
||||
|
||||
@@ -225,6 +240,14 @@ func NewResourceServer(opts ResourceServerOptions) (ResourceServer, error) {
|
||||
cancel: cancel,
|
||||
}
|
||||
|
||||
if opts.Search.Resources != nil {
|
||||
var err error
|
||||
s.search, err = newSearchSupport(opts.Search, s.backend, s.blob, opts.Tracer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
@@ -235,6 +258,7 @@ type server struct {
|
||||
log *slog.Logger
|
||||
backend StorageBackend
|
||||
blob BlobSupport
|
||||
search *searchSupport
|
||||
index ResourceIndexServer
|
||||
diagnostics DiagnosticsServer
|
||||
access authz.AccessClient
|
||||
@@ -269,6 +293,11 @@ func (s *server) Init(ctx context.Context) error {
|
||||
s.initErr = s.initWatcher()
|
||||
}
|
||||
|
||||
// initialize the search index
|
||||
if s.initErr == nil && s.search != nil {
|
||||
s.initErr = s.search.init(ctx)
|
||||
}
|
||||
|
||||
if s.initErr != nil {
|
||||
s.log.Error("error initializing resource server", "error", s.initErr)
|
||||
}
|
||||
|
||||
@@ -5,15 +5,18 @@
|
||||
"name": "ae2ntrqxefvnke",
|
||||
"namespace": "default",
|
||||
"uid": "playlist-1",
|
||||
"creationTimestamp": "2024-11-01T19:42:22Z",
|
||||
"creationTimestamp": "2024-06-01T10:11:12Z",
|
||||
"annotations": {
|
||||
"grafana.app/createdBy": "user:1",
|
||||
"grafana.app/createdBy": "user:ABC",
|
||||
"grafana.app/updatedBy": "user:XYZ",
|
||||
"grafana.app/repoName": "SQL",
|
||||
"grafana.app/repoPath": "15",
|
||||
"grafana.app/repoTimestamp": "2024-11-01T19:42:22Z"
|
||||
"grafana.app/repoHash": "xyz",
|
||||
"grafana.app/updatedTimestamp": "2024-07-01T10:11:12Z"
|
||||
}
|
||||
},
|
||||
"spec": {
|
||||
"title": "test-us-playlist"
|
||||
"title": "test playlist unified storage",
|
||||
"description": "description for the test playlist"
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user