UnifiedSearch: Use ResourceIndex from dashboards apiserver (v0alpha1 only) (#96939)
This commit is contained in:
Ryan McKinley
@@ -17,151 +17,3 @@ require (
|
||||
google.golang.org/protobuf v1.34.2
|
||||
k8s.io/apimachinery v0.31.1
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/RoaringBitmap/roaring v1.9.3 // indirect
|
||||
github.com/bits-and-blooms/bitset v1.12.0 // indirect
|
||||
github.com/blevesearch/bleve_index_api v1.1.10 // indirect
|
||||
github.com/blevesearch/geo v0.1.20 // indirect
|
||||
github.com/blevesearch/go-faiss v1.0.20 // indirect
|
||||
github.com/blevesearch/go-porterstemmer v1.0.3 // indirect
|
||||
github.com/blevesearch/gtreap v0.1.1 // indirect
|
||||
github.com/blevesearch/mmap-go v1.0.4 // indirect
|
||||
github.com/blevesearch/scorch_segment_api/v2 v2.2.15 // indirect
|
||||
github.com/blevesearch/segment v0.9.1 // indirect
|
||||
github.com/blevesearch/snowballstem v0.9.0 // indirect
|
||||
github.com/blevesearch/upsidedown_store_api v1.0.2 // indirect
|
||||
github.com/blevesearch/vellum v1.0.10 // indirect
|
||||
github.com/blevesearch/zapx/v11 v11.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v12 v12.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v13 v13.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v14 v14.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v15 v15.3.13 // indirect
|
||||
github.com/blevesearch/zapx/v16 v16.1.5 // indirect
|
||||
github.com/golang/geo v0.0.0-20210211234256-740aa86cb551 // indirect
|
||||
github.com/golang/snappy v0.0.4 // indirect
|
||||
github.com/mschoch/smat v0.2.0 // indirect
|
||||
go.etcd.io/bbolt v1.3.9 // indirect
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/RoaringBitmap/roaring v1.9.3 // indirect
|
||||
github.com/bits-and-blooms/bitset v1.12.0 // indirect
|
||||
github.com/blevesearch/bleve_index_api v1.1.10 // indirect
|
||||
github.com/blevesearch/geo v0.1.20 // indirect
|
||||
github.com/blevesearch/go-faiss v1.0.20 // indirect
|
||||
github.com/blevesearch/go-porterstemmer v1.0.3 // indirect
|
||||
github.com/blevesearch/gtreap v0.1.1 // indirect
|
||||
github.com/blevesearch/mmap-go v1.0.4 // indirect
|
||||
github.com/blevesearch/scorch_segment_api/v2 v2.2.15 // indirect
|
||||
github.com/blevesearch/segment v0.9.1 // indirect
|
||||
github.com/blevesearch/snowballstem v0.9.0 // indirect
|
||||
github.com/blevesearch/upsidedown_store_api v1.0.2 // indirect
|
||||
github.com/blevesearch/vellum v1.0.10 // indirect
|
||||
github.com/blevesearch/zapx/v11 v11.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v12 v12.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v13 v13.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v14 v14.3.10 // indirect
|
||||
github.com/blevesearch/zapx/v15 v15.3.13 // indirect
|
||||
github.com/blevesearch/zapx/v16 v16.1.5 // indirect
|
||||
github.com/golang/geo v0.0.0-20210211234256-740aa86cb551 // indirect
|
||||
github.com/golang/snappy v0.0.4 // indirect
|
||||
github.com/mschoch/smat v0.2.0 // indirect
|
||||
go.etcd.io/bbolt v1.3.9 // indirect
|
||||
)
|
||||
|
||||
require (
|
||||
cloud.google.com/go v0.115.0 // indirect
|
||||
cloud.google.com/go/auth v0.8.1 // indirect
|
||||
cloud.google.com/go/auth/oauth2adapt v0.2.4 // indirect
|
||||
cloud.google.com/go/compute/metadata v0.5.0 // indirect
|
||||
cloud.google.com/go/iam v1.1.13 // indirect
|
||||
cloud.google.com/go/storage v1.43.0 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.14.0 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.7.0 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.3.2 // indirect
|
||||
github.com/Azure/go-autorest v14.2.0+incompatible // indirect
|
||||
github.com/Azure/go-autorest/autorest/to v0.4.0 // indirect
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.2.2 // indirect
|
||||
github.com/aws/aws-sdk-go v1.55.5 // indirect
|
||||
github.com/aws/aws-sdk-go-v2 v1.30.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/config v1.27.27 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/credentials v1.17.27 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.11 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.10 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.15 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.3.17 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.17.15 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/s3 v1.58.3 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sso v1.22.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4 // indirect
|
||||
github.com/aws/aws-sdk-go-v2/service/sts v1.30.3 // indirect
|
||||
github.com/aws/smithy-go v1.20.3 // indirect
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/blevesearch/bleve/v2 v2.4.2
|
||||
github.com/bufbuild/protocompile v0.4.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
|
||||
github.com/felixge/httpsnoop v1.0.4 // indirect
|
||||
github.com/fxamacker/cbor/v2 v2.7.0 // indirect
|
||||
github.com/go-jose/go-jose/v3 v3.0.3 // indirect
|
||||
github.com/go-logr/logr v1.4.2 // indirect
|
||||
github.com/go-logr/stdr v1.2.2 // indirect
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/golang-jwt/jwt/v5 v5.2.1 // indirect
|
||||
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
|
||||
github.com/golang/protobuf v1.5.4 // indirect
|
||||
github.com/google/gofuzz v1.2.0 // indirect
|
||||
github.com/google/s2a-go v0.1.8 // indirect
|
||||
github.com/google/wire v0.6.0 // indirect
|
||||
github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/googleapis/gax-go/v2 v2.13.0 // indirect
|
||||
github.com/jhump/protoreflect v1.15.1 // indirect
|
||||
github.com/jmespath/go-jmespath v0.4.0 // indirect
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/kylelemons/godebug v1.1.0 // indirect
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
|
||||
github.com/modern-go/reflect2 v1.0.2 // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
|
||||
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
|
||||
github.com/prometheus/client_model v0.6.1 // indirect
|
||||
github.com/prometheus/common v0.55.0 // indirect
|
||||
github.com/prometheus/procfs v0.15.1 // indirect
|
||||
github.com/x448/float16 v0.8.4 // indirect
|
||||
go.opencensus.io v0.24.0 // indirect
|
||||
go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.53.0 // indirect
|
||||
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.53.0 // indirect
|
||||
go.opentelemetry.io/otel v1.29.0 // indirect
|
||||
go.opentelemetry.io/otel/metric v1.29.0 // indirect
|
||||
go.opentelemetry.io/otel/sdk v1.29.0 // indirect
|
||||
golang.org/x/crypto v0.27.0 // indirect
|
||||
golang.org/x/net v0.29.0 // indirect
|
||||
golang.org/x/oauth2 v0.23.0 // indirect
|
||||
golang.org/x/sync v0.8.0 // indirect
|
||||
golang.org/x/sys v0.25.0 // indirect
|
||||
golang.org/x/text v0.18.0 // indirect
|
||||
golang.org/x/time v0.6.0 // indirect
|
||||
golang.org/x/xerrors v0.0.0-20240716161551-93cc26a95ae9 // indirect
|
||||
google.golang.org/api v0.191.0 // indirect
|
||||
google.golang.org/genproto v0.0.0-20240812133136-8ffd90a71988 // indirect
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20240822170219-fc7c04adadcd // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20240822170219-fc7c04adadcd // indirect
|
||||
gopkg.in/inf.v0 v0.9.1 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
k8s.io/apiserver v0.31.1 // indirect
|
||||
k8s.io/klog/v2 v2.130.1 // indirect
|
||||
k8s.io/utils v0.0.0-20240711033017-18e509b52bc8 // indirect
|
||||
sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
|
||||
sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
|
||||
)
|
||||
|
||||
@@ -1,578 +0,0 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"context"
|
||||
golog "log"
|
||||
"path/filepath"
|
||||
reflect "reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/blevesearch/bleve/v2"
|
||||
"github.com/blevesearch/bleve/v2/search"
|
||||
"github.com/google/uuid"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
const tracingPrexfixIndex = "unified_storage.index."
|
||||
const specFieldPrefix = "Spec."
|
||||
const descendingPrefix = "-"
|
||||
|
||||
type Shard struct {
|
||||
index bleve.Index
|
||||
path string
|
||||
batch *bleve.Batch
|
||||
}
|
||||
|
||||
type Opts struct {
|
||||
Workers int // This controls how many goroutines are used to index objects
|
||||
BatchSize int // This is the batch size for how many objects to add to the index at once
|
||||
ListLimit int // This is how big the List page size is. If the response size is too large, the number of items will be limited by the server.
|
||||
IndexDir string // The directory where the indexes for each tenant are stored
|
||||
}
|
||||
|
||||
type Index struct {
|
||||
shardMutex sync.RWMutex
|
||||
shards map[string]*Shard
|
||||
opts Opts
|
||||
s *server
|
||||
log log.Logger
|
||||
tracer tracing.Tracer
|
||||
}
|
||||
|
||||
func NewIndex(s *server, opts Opts, tracer tracing.Tracer) *Index {
|
||||
return &Index{
|
||||
shardMutex: sync.RWMutex{},
|
||||
s: s,
|
||||
opts: opts,
|
||||
shards: make(map[string]*Shard),
|
||||
log: log.New("unifiedstorage.search.index"),
|
||||
tracer: tracer,
|
||||
}
|
||||
}
|
||||
|
||||
// IndexBatches goes through all the shards and indexes their batches if they are large enough
|
||||
func (i *Index) IndexBatches(ctx context.Context, maxSize int, tenants []string) error {
|
||||
_, span := i.tracer.Start(ctx, tracingPrexfixIndex+"IndexBatches")
|
||||
defer span.End()
|
||||
|
||||
group := errgroup.Group{}
|
||||
group.SetLimit(i.opts.Workers)
|
||||
totalBatchesIndexed := 0
|
||||
|
||||
for _, tenant := range tenants {
|
||||
shard, err := i.getShard(tenant)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Index the batch if it is large enough
|
||||
if shard.batch.Size() >= maxSize {
|
||||
totalBatchesIndexed++
|
||||
group.Go(func() error {
|
||||
i.log.Debug("indexing batch for shard", "tenant", tenant, "size", shard.batch.Size())
|
||||
err = shard.index.Batch(shard.batch)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
shard.batch.Reset()
|
||||
return nil
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
err := group.Wait()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
span.AddEvent("batches indexed", trace.WithAttributes(attribute.Int("batches_indexed", totalBatchesIndexed)))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddToBatches adds resources to their respective shard's batch
|
||||
// returns a list of tenants that have changes
|
||||
func (i *Index) AddToBatches(ctx context.Context, list *ListResponse) ([]string, error) {
|
||||
_, span := i.tracer.Start(ctx, tracingPrexfixIndex+"AddToBatches")
|
||||
defer span.End()
|
||||
|
||||
tenantsWithChanges := make(map[string]bool)
|
||||
for _, obj := range list.Items {
|
||||
// Transform the raw resource into a more generic indexable resource
|
||||
res, err := NewIndexedResource(obj.Value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
shard, err := i.getShard(res.Namespace)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
i.log.Debug("indexing resource in batch", "batch_count", len(list.Items), "kind", res.Kind, "tenant", res.Namespace)
|
||||
|
||||
err = shard.batch.Index(res.Uid, res)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if _, ok := tenantsWithChanges[res.Namespace]; !ok {
|
||||
tenantsWithChanges[res.Namespace] = true
|
||||
}
|
||||
}
|
||||
|
||||
tenants := make([]string, 0, len(tenantsWithChanges))
|
||||
for tenant := range tenantsWithChanges {
|
||||
tenants = append(tenants, tenant)
|
||||
}
|
||||
|
||||
return tenants, nil
|
||||
}
|
||||
|
||||
func (i *Index) Init(ctx context.Context) error {
|
||||
logger := i.log.FromContext(ctx)
|
||||
ctx, span := i.tracer.Start(ctx, tracingPrexfixIndex+"Init")
|
||||
defer span.End()
|
||||
|
||||
start := time.Now().Unix()
|
||||
group := errgroup.Group{}
|
||||
group.SetLimit(i.opts.Workers)
|
||||
|
||||
totalObjects := 0
|
||||
// Get all tenants currently in Unified Storage
|
||||
tenants, err := i.s.backend.Namespaces(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, tenant := range tenants {
|
||||
group.Go(func() error {
|
||||
logger.Info("initializing index for tenant", "tenant", tenant)
|
||||
objs, err := i.InitForTenant(ctx, tenant)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
totalObjects += objs
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
err = group.Wait()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
//index all remaining batches for all tenants
|
||||
logger.Info("indexing remaining batches", "shards", len(i.shards))
|
||||
err = i.IndexBatches(ctx, 1, i.allTenants())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
end := time.Now().Unix()
|
||||
totalDocCount := getTotalDocCount(i)
|
||||
logger.Info("Initial indexing finished", "seconds", float64(end-start), "objs_fetched", totalObjects, "objs_indexed", totalDocCount)
|
||||
span.AddEvent(
|
||||
"indexing finished",
|
||||
trace.WithAttributes(attribute.Int64("objects_indexed", int64(totalDocCount))),
|
||||
trace.WithAttributes(attribute.Int64("objects_fetched", int64(totalObjects))),
|
||||
)
|
||||
if IndexServerMetrics != nil {
|
||||
IndexServerMetrics.IndexCreationTime.WithLabelValues().Observe(float64(end - start))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (i *Index) InitForTenant(ctx context.Context, namespace string) (int, error) {
|
||||
ctx, span := i.tracer.Start(ctx, tracingPrexfixIndex+"InitForTenant")
|
||||
defer span.End()
|
||||
logger := i.log.FromContext(ctx)
|
||||
|
||||
resourceTypes := fetchResourceTypes()
|
||||
totalObjectsFetched := 0
|
||||
for _, rt := range resourceTypes {
|
||||
logger.Debug("indexing resource", "kind", rt.Kind, "list_limit", i.opts.ListLimit, "batch_size", i.opts.BatchSize, "workers", i.opts.Workers, "namespace", namespace)
|
||||
r := &ListRequest{Options: rt.ListOptions, Limit: int64(i.opts.ListLimit)}
|
||||
r.Options.Key.Namespace = namespace // scope the list to a tenant or this will take forever when US has 1M+ resources
|
||||
|
||||
// Paginate through the list of resources and index each page
|
||||
for {
|
||||
logger.Debug("fetching resource list", "kind", rt.Kind, "namespace", namespace)
|
||||
list, err := i.s.List(ctx, r)
|
||||
if err != nil {
|
||||
return totalObjectsFetched, err
|
||||
}
|
||||
|
||||
// Record the number of objects indexed for the kind
|
||||
IndexServerMetrics.IndexedKinds.WithLabelValues(rt.Kind).Add(float64(len(list.Items)))
|
||||
|
||||
totalObjectsFetched += len(list.Items)
|
||||
|
||||
logger.Debug("indexing batch", "kind", rt.Kind, "count", len(list.Items), "namespace", namespace)
|
||||
//add changes to batches for shards with changes in the List
|
||||
err = i.writeBatch(ctx, list)
|
||||
if err != nil {
|
||||
return totalObjectsFetched, err
|
||||
}
|
||||
|
||||
if list.NextPageToken == "" {
|
||||
break
|
||||
}
|
||||
|
||||
r.NextPageToken = list.NextPageToken
|
||||
}
|
||||
}
|
||||
|
||||
span.AddEvent(
|
||||
"indexing finished for tenant",
|
||||
trace.WithAttributes(attribute.Int64("objects_indexed", int64(totalObjectsFetched))),
|
||||
trace.WithAttributes(attribute.String("tenant", namespace)),
|
||||
)
|
||||
|
||||
return totalObjectsFetched, nil
|
||||
}
|
||||
|
||||
func (i *Index) writeBatch(ctx context.Context, list *ListResponse) error {
|
||||
tenants, err := i.AddToBatches(ctx, list)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Index the batches for tenants with changes if the batch is large enough
|
||||
err = i.IndexBatches(ctx, i.opts.BatchSize, tenants)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (i *Index) Index(ctx context.Context, data *Data) error {
|
||||
ctx, span := i.tracer.Start(ctx, tracingPrexfixIndex+"Index")
|
||||
defer span.End()
|
||||
logger := i.log.FromContext(ctx)
|
||||
|
||||
// Transform the raw resource into a more generic indexable resource
|
||||
res, err := NewIndexedResource(data.Value.Value)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tenant := res.Namespace
|
||||
logger.Debug("indexing resource for tenant", "res", string(data.Value.Value), "tenant", tenant)
|
||||
|
||||
// if tenant doesn't exist, they may have been created during initial indexing
|
||||
_, ok := i.shards[tenant]
|
||||
if !ok {
|
||||
i.log.Info("tenant not found, initializing their index", "tenant", tenant)
|
||||
_, err = i.InitForTenant(ctx, tenant)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
shard, err := i.getShard(tenant)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = shard.index.Index(res.Uid, res)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
//record the kind of resource that was indexed
|
||||
IndexServerMetrics.IndexedKinds.WithLabelValues(res.Kind).Inc()
|
||||
|
||||
// record latency from when event was created to when it was indexed
|
||||
latencySeconds := float64(time.Now().UnixMicro()-data.Value.ResourceVersion) / 1e6
|
||||
if latencySeconds > 5 {
|
||||
logger.Warn("high index latency", "latency", latencySeconds)
|
||||
}
|
||||
if IndexServerMetrics != nil {
|
||||
IndexServerMetrics.IndexLatency.WithLabelValues(data.Key.Resource).Observe(latencySeconds)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (i *Index) Delete(ctx context.Context, uid string, key *ResourceKey) error {
|
||||
_, span := i.tracer.Start(ctx, tracingPrexfixIndex+"Delete")
|
||||
defer span.End()
|
||||
|
||||
shard, err := i.getShard(key.Namespace)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = shard.index.Delete(uid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
IndexServerMetrics.IndexedKinds.WithLabelValues(key.Resource).Dec()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (i *Index) Search(ctx context.Context, request *SearchRequest) (*IndexResults, error) {
|
||||
ctx, span := i.tracer.Start(ctx, tracingPrexfixIndex+"Search")
|
||||
defer span.End()
|
||||
logger := i.log.FromContext(ctx)
|
||||
|
||||
if request.Tenant == "" {
|
||||
request.Tenant = "default"
|
||||
}
|
||||
shard, err := i.getShard(request.Tenant)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
docCount, err := shard.index.DocCount()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
logger.Info("got index for tenant", "tenant", request.Tenant, "docCount", docCount)
|
||||
|
||||
fields, _ := shard.index.Fields()
|
||||
logger.Debug("indexed fields", "fields", fields)
|
||||
|
||||
// use 10 as a default limit for now
|
||||
if request.Limit <= 0 {
|
||||
request.Limit = 10
|
||||
}
|
||||
|
||||
textQuery := bleve.NewQueryStringQuery(request.Query)
|
||||
query := bleve.NewConjunctionQuery(textQuery)
|
||||
|
||||
if len(request.Kind) > 0 {
|
||||
// apply OR condition filter for each kind ( dashboard, folder, etc )
|
||||
orQuery := bleve.NewDisjunctionQuery()
|
||||
for _, term := range request.Kind {
|
||||
termQuery := bleve.NewTermQuery(term)
|
||||
orQuery.AddQuery(termQuery)
|
||||
}
|
||||
query.AddQuery(orQuery)
|
||||
}
|
||||
|
||||
if len(request.Filters) > 0 {
|
||||
orQuery := bleve.NewDisjunctionQuery()
|
||||
for _, filter := range request.Filters {
|
||||
matchQuery := bleve.NewMatchQuery(filter)
|
||||
orQuery.AddQuery(matchQuery)
|
||||
}
|
||||
query.AddQuery(orQuery)
|
||||
}
|
||||
|
||||
req := bleve.NewSearchRequest(query)
|
||||
if len(request.SortBy) > 0 {
|
||||
sorting := getSortFields(request)
|
||||
req.SortBy(sorting)
|
||||
}
|
||||
|
||||
for _, group := range request.GroupBy {
|
||||
facet := bleve.NewFacetRequest(specFieldPrefix+group.Name, int(group.Limit))
|
||||
req.AddFacet(group.Name+"_facet", facet)
|
||||
}
|
||||
|
||||
req.From = int(request.Offset)
|
||||
req.Size = int(request.Limit)
|
||||
|
||||
req.Fields = []string{"*"} // return all indexed fields in search results
|
||||
|
||||
logger.Info("searching index", "query", request.Query, "tenant", request.Tenant)
|
||||
res, err := shard.index.Search(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
hits := res.Hits
|
||||
|
||||
logger.Info("got search results", "hits", hits)
|
||||
|
||||
results := make([]IndexedResource, len(hits))
|
||||
for resKey, hit := range hits {
|
||||
ir := IndexedResource{}.FromSearchHit(hit)
|
||||
results[resKey] = ir
|
||||
}
|
||||
|
||||
groups := []*Group{}
|
||||
for _, group := range request.GroupBy {
|
||||
groupByFacet := res.Facets[group.Name+"_facet"]
|
||||
terms := getTermFacets(groupByFacet.Terms)
|
||||
for _, term := range terms {
|
||||
groups = append(groups, &Group{Name: term.Term, Count: int64(term.Count)})
|
||||
}
|
||||
}
|
||||
|
||||
return &IndexResults{Values: results, Groups: groups}, nil
|
||||
}
|
||||
|
||||
// Count returns the total doc count
|
||||
func (i *Index) Count() (int, error) {
|
||||
total := 0
|
||||
for _, shard := range i.shards {
|
||||
count, err := shard.index.DocCount()
|
||||
if err != nil {
|
||||
i.log.Error("failed to get doc count", "error", err)
|
||||
}
|
||||
total += int(count)
|
||||
}
|
||||
return total, nil
|
||||
}
|
||||
|
||||
// allTenants returns a list of all tenants in the index
|
||||
func (i *Index) allTenants() []string {
|
||||
tenants := make([]string, 0, len(i.shards))
|
||||
for tenant := range i.shards {
|
||||
tenants = append(tenants, tenant)
|
||||
}
|
||||
return tenants
|
||||
}
|
||||
|
||||
func (i *Index) getShard(tenant string) (*Shard, error) {
|
||||
i.shardMutex.Lock()
|
||||
defer i.shardMutex.Unlock()
|
||||
|
||||
shard, ok := i.shards[tenant]
|
||||
if ok {
|
||||
return shard, nil
|
||||
}
|
||||
|
||||
index, path, err := i.createIndex()
|
||||
if err != nil {
|
||||
return &Shard{}, err
|
||||
}
|
||||
|
||||
shard = &Shard{
|
||||
index: index,
|
||||
path: path,
|
||||
batch: index.NewBatch(),
|
||||
}
|
||||
i.shards[tenant] = shard
|
||||
|
||||
return shard, nil
|
||||
}
|
||||
|
||||
func (i *Index) createIndex() (bleve.Index, string, error) {
|
||||
if i.opts.IndexDir == "" {
|
||||
return createInMemoryIndex()
|
||||
}
|
||||
return createFileIndex(i.opts.IndexDir)
|
||||
}
|
||||
|
||||
var mappings = createIndexMappings()
|
||||
|
||||
// less memory intensive alternative for larger indexes with less tenants (on-prem)
|
||||
func createFileIndex(path string) (bleve.Index, string, error) {
|
||||
indexPath := filepath.Join(path, uuid.New().String())
|
||||
index, err := bleve.New(indexPath, mappings)
|
||||
if err != nil {
|
||||
golog.Fatalf("Failed to create index: %v", err)
|
||||
}
|
||||
return index, indexPath, err
|
||||
}
|
||||
|
||||
// faster indexing when there are many tenants with smaller batches (cloud)
|
||||
func createInMemoryIndex() (bleve.Index, string, error) {
|
||||
index, err := bleve.NewMemOnly(mappings)
|
||||
return index, "", err
|
||||
}
|
||||
|
||||
type IndexerListOptions struct {
|
||||
*ListOptions
|
||||
Kind string
|
||||
}
|
||||
|
||||
// TODO - fetch from api
|
||||
// Folders need to be indexed first as dashboards depend on them to be indexed already.
|
||||
func fetchResourceTypes() []*IndexerListOptions {
|
||||
return []*IndexerListOptions{
|
||||
{
|
||||
ListOptions: &ListOptions{
|
||||
Key: &ResourceKey{
|
||||
Group: "folder.grafana.app",
|
||||
Resource: "folders",
|
||||
},
|
||||
},
|
||||
Kind: "Folder",
|
||||
},
|
||||
{
|
||||
ListOptions: &ListOptions{
|
||||
Key: &ResourceKey{
|
||||
Group: "playlist.grafana.app",
|
||||
Resource: "playlists",
|
||||
},
|
||||
},
|
||||
Kind: "Playlist",
|
||||
},
|
||||
{
|
||||
ListOptions: &ListOptions{
|
||||
Key: &ResourceKey{
|
||||
Group: "dashboard.grafana.app",
|
||||
Resource: "dashboards",
|
||||
},
|
||||
},
|
||||
Kind: "Dashboard",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getSortFields(request *SearchRequest) []string {
|
||||
sorting := make([]string, 0, len(request.SortBy))
|
||||
for _, sort := range request.SortBy {
|
||||
if IsSpecField(sort) {
|
||||
descending := strings.HasPrefix(sort, descendingPrefix)
|
||||
sort = strings.TrimPrefix(sort, descendingPrefix)
|
||||
sortOrder := ""
|
||||
if descending {
|
||||
sortOrder = descendingPrefix
|
||||
}
|
||||
sorting = append(sorting, sortOrder+specFieldPrefix+sort)
|
||||
continue
|
||||
}
|
||||
sorting = append(sorting, sort)
|
||||
}
|
||||
return sorting
|
||||
}
|
||||
|
||||
func getTermFacets(f *search.TermFacets) []*search.TermFacet {
|
||||
e := reflect.ValueOf(f).Elem()
|
||||
if e.Kind() != reflect.Struct {
|
||||
return []*search.TermFacet{}
|
||||
}
|
||||
// workaround - this field is private, so we need to use reflection to access it
|
||||
// TODO - fork bleve and create a pr to make this field accessible
|
||||
v := e.FieldByName("termLookup")
|
||||
if v.Kind() != reflect.Map {
|
||||
return []*search.TermFacet{}
|
||||
}
|
||||
|
||||
terms := []*search.TermFacet{}
|
||||
termsRange := v.MapRange()
|
||||
for termsRange.Next() {
|
||||
value := termsRange.Value()
|
||||
// facet value is *search.TermFacet
|
||||
if value.Kind() == reflect.Pointer {
|
||||
val := value.Elem()
|
||||
if val.Kind() == reflect.Struct {
|
||||
group := newTerm(val)
|
||||
terms = append(terms, group)
|
||||
}
|
||||
}
|
||||
}
|
||||
return terms
|
||||
}
|
||||
|
||||
func newTerm(val reflect.Value) *search.TermFacet {
|
||||
term := &search.TermFacet{}
|
||||
|
||||
for i := 0; i < val.NumField(); i++ {
|
||||
field := val.Field(i)
|
||||
if field.Kind() == reflect.String {
|
||||
term.Term = field.String()
|
||||
}
|
||||
if field.Kind() == reflect.Int {
|
||||
term.Count = int(field.Int())
|
||||
}
|
||||
}
|
||||
return term
|
||||
}
|
||||
@@ -1,253 +0,0 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/blevesearch/bleve/v2"
|
||||
"github.com/blevesearch/bleve/v2/mapping"
|
||||
"github.com/blevesearch/bleve/v2/search"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
)
|
||||
|
||||
type IndexedResource struct {
|
||||
Uid string
|
||||
Group string
|
||||
Namespace string
|
||||
Kind string
|
||||
Name string
|
||||
Title string
|
||||
CreatedAt string
|
||||
CreatedBy string
|
||||
UpdatedAt string
|
||||
UpdatedBy string
|
||||
FolderId string
|
||||
Spec map[string]any
|
||||
}
|
||||
|
||||
type IndexResults struct {
|
||||
Values []IndexedResource
|
||||
Groups []*Group
|
||||
}
|
||||
|
||||
func (ir IndexedResource) FromSearchHit(hit *search.DocumentMatch) IndexedResource {
|
||||
ir.Uid = fieldValue("Uid", hit)
|
||||
ir.Kind = fieldValue("Kind", hit)
|
||||
ir.Name = fieldValue("Name", hit)
|
||||
ir.Namespace = fieldValue("Namespace", hit)
|
||||
ir.Group = fieldValue("Group", hit)
|
||||
ir.CreatedAt = fieldValue("CreatedAt", hit)
|
||||
ir.CreatedBy = fieldValue("CreatedBy", hit)
|
||||
ir.UpdatedAt = fieldValue("UpdatedAt", hit)
|
||||
ir.UpdatedBy = fieldValue("UpdatedBy", hit)
|
||||
ir.Title = fieldValue("Title", hit)
|
||||
|
||||
// add indexed spec fields to search results
|
||||
specResult := map[string]any{}
|
||||
for k, v := range hit.Fields {
|
||||
if strings.HasPrefix(k, "Spec.") {
|
||||
specKey := strings.TrimPrefix(k, "Spec.")
|
||||
specResult[specKey] = v
|
||||
}
|
||||
ir.Spec = specResult
|
||||
}
|
||||
|
||||
return ir
|
||||
}
|
||||
|
||||
func fieldValue(field string, hit *search.DocumentMatch) string {
|
||||
if val, ok := hit.Fields[field]; ok {
|
||||
return val.(string)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// NewIndexedResource creates a new IndexedResource from a raw resource.
|
||||
// rawResource is the raw json for the resource from unified storage.
|
||||
func NewIndexedResource(rawResource []byte) (*IndexedResource, error) {
|
||||
k8sObj := unstructured.Unstructured{}
|
||||
err := k8sObj.UnmarshalJSON(rawResource)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
meta, err := utils.MetaAccessor(&k8sObj)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ir := &IndexedResource{}
|
||||
ir.Uid = string(meta.GetUID())
|
||||
ir.Name = meta.GetName()
|
||||
ir.Title = meta.FindTitle("")
|
||||
ir.Namespace = meta.GetNamespace()
|
||||
ir.Group = meta.GetGroupVersionKind().Group
|
||||
ir.Kind = meta.GetGroupVersionKind().Kind
|
||||
ir.CreatedAt = meta.GetCreationTimestamp().Time.Format("2006-01-02T15:04:05Z")
|
||||
ir.CreatedBy = meta.GetCreatedBy()
|
||||
updatedAt, err := meta.GetUpdatedTimestamp()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if updatedAt != nil {
|
||||
ir.UpdatedAt = updatedAt.Format("2006-01-02T15:04:05Z")
|
||||
} else {
|
||||
ir.UpdatedAt = ir.CreatedAt
|
||||
}
|
||||
ir.UpdatedBy = meta.GetUpdatedBy()
|
||||
spec, err := meta.GetSpec()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
specValues, ok := spec.(map[string]any)
|
||||
if ok {
|
||||
ir.Spec = specValues
|
||||
}
|
||||
|
||||
return ir, nil
|
||||
}
|
||||
|
||||
func createIndexMappings() *mapping.IndexMappingImpl {
|
||||
// Create the index mapping
|
||||
indexMapping := bleve.NewIndexMapping()
|
||||
// Create an individual index mapping for each kind
|
||||
indexMapping.TypeField = "Kind"
|
||||
|
||||
// for all kinds, create their index mappings
|
||||
for k := range getSpecObjectMappings() {
|
||||
objMapping := createIndexMappingForKind(k)
|
||||
indexMapping.AddDocumentMapping(k, objMapping)
|
||||
}
|
||||
|
||||
return indexMapping
|
||||
}
|
||||
|
||||
func createIndexMappingForKind(resourceKind string) *mapping.DocumentMapping {
|
||||
// create mappings for top level fields
|
||||
baseFields := map[string]*mapping.FieldMapping{
|
||||
"Uid": bleve.NewTextFieldMapping(),
|
||||
"Group": bleve.NewTextFieldMapping(),
|
||||
"Namespace": bleve.NewTextFieldMapping(),
|
||||
"Kind": bleve.NewTextFieldMapping(),
|
||||
"Name": bleve.NewTextFieldMapping(),
|
||||
"Title": bleve.NewTextFieldMapping(),
|
||||
"CreatedAt": bleve.NewDateTimeFieldMapping(),
|
||||
"CreatedBy": bleve.NewTextFieldMapping(),
|
||||
"UpdatedAt": bleve.NewDateTimeFieldMapping(),
|
||||
"UpdatedBy": bleve.NewTextFieldMapping(),
|
||||
"FolderId": bleve.NewTextFieldMapping(),
|
||||
}
|
||||
|
||||
// Spec is different for all resources, so we need to generate the spec mapping based on the kind
|
||||
specMapping := createSpecObjectMapping(resourceKind)
|
||||
|
||||
objectMapping := bleve.NewDocumentMapping()
|
||||
objectMapping.Dynamic = false // only map fields that we have explicitly defined
|
||||
|
||||
// map spec
|
||||
objectMapping.AddSubDocumentMapping("Spec", specMapping)
|
||||
|
||||
// map top level fields
|
||||
for k, v := range baseFields {
|
||||
objectMapping.AddFieldMappingsAt(k, v)
|
||||
}
|
||||
|
||||
return objectMapping
|
||||
}
|
||||
|
||||
type SpecFieldMapping struct {
|
||||
Field string
|
||||
Type string
|
||||
}
|
||||
|
||||
// Right now we are hardcoding which spec fields to index for each kind
|
||||
// In the future, which fields to index will be defined on the resources themselves by their owners.
|
||||
func getSpecObjectMappings() map[string][]SpecFieldMapping {
|
||||
return specMappings
|
||||
}
|
||||
|
||||
// Generate the spec field mapping for a given kind
|
||||
func createSpecObjectMapping(kind string) *mapping.DocumentMapping {
|
||||
specMapping := bleve.NewDocumentMapping()
|
||||
specMapping.Dynamic = false
|
||||
|
||||
// get the fields to index for the kind
|
||||
mappings := getSpecObjectMappings()[kind]
|
||||
|
||||
for _, m := range mappings {
|
||||
fieldName := m.Field
|
||||
fieldType := m.Type
|
||||
|
||||
// Create a field mapping based on field type
|
||||
switch fieldType {
|
||||
case "string", "string[]":
|
||||
specMapping.AddFieldMappingsAt(fieldName, bleve.NewTextFieldMapping())
|
||||
case "int", "int64", "float64":
|
||||
specMapping.AddFieldMappingsAt(fieldName, bleve.NewNumericFieldMapping())
|
||||
case "bool":
|
||||
specMapping.AddFieldMappingsAt(fieldName, bleve.NewBooleanFieldMapping())
|
||||
case "time":
|
||||
specMapping.AddFieldMappingsAt(fieldName, bleve.NewDateTimeFieldMapping())
|
||||
default:
|
||||
// TODO support indexing arrays and nested fields
|
||||
// We are only indexing top level string,int, and bool fields within spec for now. Arrays or nested fields are not yet supported.
|
||||
}
|
||||
}
|
||||
|
||||
return specMapping
|
||||
}
|
||||
|
||||
func IsSpecField(field string) bool {
|
||||
field = strings.TrimPrefix(field, "-")
|
||||
_, ok := specFields[field]
|
||||
return ok
|
||||
}
|
||||
|
||||
var specFields = mapSpecFields()
|
||||
|
||||
func mapSpecFields() map[string]bool {
|
||||
fields := map[string]bool{}
|
||||
for _, mappings := range specMappings {
|
||||
for _, m := range mappings {
|
||||
fields[m.Field] = true
|
||||
}
|
||||
}
|
||||
return fields
|
||||
}
|
||||
|
||||
var specMappings = map[string][]SpecFieldMapping{
|
||||
"Playlist": {
|
||||
{
|
||||
Field: "interval",
|
||||
Type: "string",
|
||||
},
|
||||
{
|
||||
Field: "title",
|
||||
Type: "string",
|
||||
},
|
||||
},
|
||||
"Folder": {
|
||||
{
|
||||
Field: "title",
|
||||
Type: "string",
|
||||
},
|
||||
{
|
||||
Field: "description",
|
||||
Type: "string",
|
||||
},
|
||||
},
|
||||
"Dashboard": {
|
||||
{
|
||||
Field: "title",
|
||||
Type: "string",
|
||||
},
|
||||
{
|
||||
Field: "description",
|
||||
Type: "string",
|
||||
},
|
||||
{
|
||||
Field: "tags",
|
||||
Type: "string[]",
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -1,126 +0,0 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/dskit/instrument"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
var (
|
||||
onceIndex sync.Once
|
||||
IndexServerMetrics *IndexMetrics
|
||||
)
|
||||
|
||||
type IndexMetrics struct {
|
||||
IndexDir string
|
||||
IndexServer *IndexServer
|
||||
|
||||
// metrics
|
||||
IndexLatency *prometheus.HistogramVec
|
||||
IndexSize prometheus.Gauge
|
||||
IndexedDocs prometheus.Gauge
|
||||
IndexedKinds *prometheus.GaugeVec
|
||||
IndexCreationTime *prometheus.HistogramVec
|
||||
}
|
||||
|
||||
var IndexCreationBuckets = []float64{1, 5, 10, 25, 50, 75, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000}
|
||||
|
||||
func NewIndexMetrics(indexDir string, indexServer *IndexServer) *IndexMetrics {
|
||||
onceIndex.Do(func() {
|
||||
IndexServerMetrics = &IndexMetrics{
|
||||
IndexDir: indexDir,
|
||||
IndexServer: indexServer,
|
||||
IndexLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Namespace: "index_server",
|
||||
Name: "index_latency_seconds",
|
||||
Help: "Time (in seconds) until index is updated with new event",
|
||||
Buckets: instrument.DefBuckets,
|
||||
NativeHistogramBucketFactor: 1.1, // enable native histograms
|
||||
NativeHistogramMaxBucketNumber: 160,
|
||||
NativeHistogramMinResetDuration: time.Hour,
|
||||
}, []string{"resource"}),
|
||||
IndexSize: prometheus.NewGauge(prometheus.GaugeOpts{
|
||||
Namespace: "index_server",
|
||||
Name: "index_size",
|
||||
Help: "Size of the index in bytes",
|
||||
}),
|
||||
IndexedDocs: prometheus.NewGauge(prometheus.GaugeOpts{
|
||||
Namespace: "index_server",
|
||||
Name: "indexed_docs",
|
||||
Help: "Number of indexed documents by resource",
|
||||
}),
|
||||
IndexedKinds: prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
||||
Namespace: "index_server",
|
||||
Name: "indexed_kinds",
|
||||
Help: "Number of indexed documents by kind",
|
||||
}, []string{"kind"}),
|
||||
IndexCreationTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Namespace: "index_server",
|
||||
Name: "index_creation_time_seconds",
|
||||
Help: "Time (in seconds) it takes until index is created",
|
||||
Buckets: IndexCreationBuckets,
|
||||
NativeHistogramBucketFactor: 1.1, // enable native histograms
|
||||
NativeHistogramMaxBucketNumber: 160,
|
||||
NativeHistogramMinResetDuration: time.Hour,
|
||||
}, []string{}),
|
||||
}
|
||||
})
|
||||
|
||||
return IndexServerMetrics
|
||||
}
|
||||
|
||||
func (s *IndexMetrics) Collect(ch chan<- prometheus.Metric) {
|
||||
s.IndexLatency.Collect(ch)
|
||||
s.IndexCreationTime.Collect(ch)
|
||||
s.IndexedKinds.Collect(ch)
|
||||
|
||||
// collect index size
|
||||
totalSize, err := getTotalIndexSize(s.IndexDir)
|
||||
if err == nil {
|
||||
s.IndexSize.Set(float64(totalSize))
|
||||
s.IndexSize.Collect(ch)
|
||||
}
|
||||
|
||||
// collect index docs
|
||||
s.IndexedDocs.Set(getTotalDocCount(s.IndexServer.index))
|
||||
s.IndexedDocs.Collect(ch)
|
||||
}
|
||||
|
||||
func (s *IndexMetrics) Describe(ch chan<- *prometheus.Desc) {
|
||||
s.IndexLatency.Describe(ch)
|
||||
s.IndexSize.Describe(ch)
|
||||
s.IndexedDocs.Describe(ch)
|
||||
s.IndexedKinds.Describe(ch)
|
||||
s.IndexCreationTime.Describe(ch)
|
||||
}
|
||||
|
||||
// getTotalDocCount returns the total number of documents in the index
|
||||
func getTotalDocCount(index *Index) float64 {
|
||||
count, _ := index.Count()
|
||||
return float64(count)
|
||||
}
|
||||
|
||||
// getTotalIndexSize returns the total size of the index directory when using a file-based index
|
||||
func getTotalIndexSize(dir string) (int64, error) {
|
||||
var totalSize int64
|
||||
|
||||
err := filepath.WalkDir(dir, func(path string, info os.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !info.IsDir() {
|
||||
fileInfo, err := info.Info()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
totalSize += fileInfo.Size()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
return totalSize, err
|
||||
}
|
||||
@@ -1,254 +0,0 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
type IndexServer struct {
|
||||
ResourceServer
|
||||
s *server
|
||||
index *Index
|
||||
ws *indexWatchServer
|
||||
log *slog.Logger
|
||||
cfg *setting.Cfg
|
||||
tracer tracing.Tracer
|
||||
}
|
||||
|
||||
const tracingPrefixIndexServer = "unified_storage.index_server."
|
||||
|
||||
func (is *IndexServer) Search(ctx context.Context, req *SearchRequest) (*SearchResponse, error) {
|
||||
ctx, span := is.tracer.Start(ctx, tracingPrefixIndexServer+"Search")
|
||||
defer span.End()
|
||||
|
||||
results, err := is.index.Search(ctx, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res := &SearchResponse{}
|
||||
for _, r := range results.Values {
|
||||
resJsonBytes, err := json.Marshal(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res.Items = append(res.Items, &ResourceWrapper{Value: resJsonBytes})
|
||||
res.Groups = results.Groups
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (is *IndexServer) History(ctx context.Context, req *HistoryRequest) (*HistoryResponse, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (is *IndexServer) Origin(ctx context.Context, req *OriginRequest) (*OriginResponse, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Load the index
|
||||
func (is *IndexServer) Load(ctx context.Context) error {
|
||||
ctx, span := is.tracer.Start(ctx, tracingPrefixIndexServer+"Load")
|
||||
defer span.End()
|
||||
|
||||
opts := Opts{
|
||||
Workers: is.cfg.IndexWorkers,
|
||||
BatchSize: is.cfg.IndexMaxBatchSize,
|
||||
ListLimit: is.cfg.IndexListLimit,
|
||||
IndexDir: is.cfg.IndexPath,
|
||||
}
|
||||
is.index = NewIndex(is.s, opts, is.tracer)
|
||||
err := is.index.Init(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Watch resources for changes and update the index
|
||||
func (is *IndexServer) Watch(ctx context.Context) error {
|
||||
rtList := fetchResourceTypes()
|
||||
for _, rt := range rtList {
|
||||
wr := &WatchRequest{
|
||||
Options: rt.ListOptions,
|
||||
}
|
||||
|
||||
go func() {
|
||||
for {
|
||||
// blocking call
|
||||
err := is.s.Watch(wr, is.ws)
|
||||
if err != nil {
|
||||
is.log.Error("Error watching resource", "error", err)
|
||||
}
|
||||
is.log.Debug("Resource watch ended. Restarting watch")
|
||||
}
|
||||
}()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Init sets the resource server on the index server
|
||||
// so we can call the resource server from the index server
|
||||
// TODO: a chicken and egg problem - index server needs the resource server but the resource server is created with the index server
|
||||
func (is *IndexServer) Init(ctx context.Context, rs *server) error {
|
||||
is.s = rs
|
||||
is.ws = &indexWatchServer{
|
||||
is: is,
|
||||
context: ctx,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewResourceIndexServer(cfg *setting.Cfg, tracer tracing.Tracer) ResourceIndexServer {
|
||||
logger := slog.Default().With("logger", "index-server")
|
||||
|
||||
indexServer := &IndexServer{
|
||||
log: logger,
|
||||
cfg: cfg,
|
||||
tracer: tracer,
|
||||
}
|
||||
|
||||
err := prometheus.Register(NewIndexMetrics(cfg.IndexPath, indexServer))
|
||||
if err != nil {
|
||||
logger.Warn("Failed to register index metrics", "error", err)
|
||||
}
|
||||
|
||||
return indexServer
|
||||
}
|
||||
|
||||
type ResourceIndexer interface {
|
||||
Index(ctx context.Context) (*Index, error)
|
||||
}
|
||||
|
||||
type indexWatchServer struct {
|
||||
grpc.ServerStream
|
||||
context context.Context
|
||||
is *IndexServer
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) Send(we *WatchEvent) error {
|
||||
if we.Type == WatchEvent_ADDED {
|
||||
return f.Add(we)
|
||||
}
|
||||
|
||||
if we.Type == WatchEvent_DELETED {
|
||||
return f.Delete(we)
|
||||
}
|
||||
|
||||
if we.Type == WatchEvent_MODIFIED {
|
||||
return f.Update(we)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) RecvMsg(m interface{}) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) SendMsg(m interface{}) error {
|
||||
return errors.New("not implemented")
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) Context() context.Context {
|
||||
if f.context == nil {
|
||||
f.context = context.Background()
|
||||
}
|
||||
return f.context
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) Index() *Index {
|
||||
return f.is.index
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) Add(we *WatchEvent) error {
|
||||
data, err := getData(we.Resource)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = f.Index().Index(f.context, data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) Delete(we *WatchEvent) error {
|
||||
rs, err := resource(we)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
data, err := getData(rs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = f.Index().Delete(f.context, data.Uid, data.Key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *indexWatchServer) Update(we *WatchEvent) error {
|
||||
rs, err := resource(we)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
data, err := getData(rs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = f.Index().Delete(f.context, data.Uid, data.Key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = f.Index().Index(f.context, data)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type Data struct {
|
||||
Key *ResourceKey
|
||||
Value *ResourceWrapper
|
||||
Uid string
|
||||
}
|
||||
|
||||
func getData(wr *WatchEvent_Resource) (*Data, error) {
|
||||
r, err := NewIndexedResource(wr.Value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
key := &ResourceKey{
|
||||
Group: r.Group,
|
||||
Resource: r.Kind, // We use Kind as resource key since watch events don't have a resource name on them
|
||||
Namespace: r.Namespace,
|
||||
Name: r.Name,
|
||||
}
|
||||
|
||||
value := &ResourceWrapper{
|
||||
ResourceVersion: wr.Version,
|
||||
Value: wr.Value,
|
||||
}
|
||||
return &Data{Key: key, Value: value, Uid: r.Uid}, nil
|
||||
}
|
||||
|
||||
func resource(we *WatchEvent) (*WatchEvent_Resource, error) {
|
||||
rs := we.Resource
|
||||
if rs == nil || len(rs.Value) == 0 {
|
||||
// for updates/deletes
|
||||
rs = we.Previous
|
||||
}
|
||||
if rs == nil || len(rs.Value) == 0 {
|
||||
return nil, errors.New("resource not found")
|
||||
}
|
||||
return rs, nil
|
||||
}
|
||||
@@ -1,281 +0,0 @@
|
||||
package resource
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.org/x/exp/rand"
|
||||
)
|
||||
|
||||
const testTenant = "default"
|
||||
|
||||
var testContext = context.Background()
|
||||
|
||||
func TestIndexDashboard(t *testing.T) {
|
||||
data := readTestData(t, "dashboard-resource.json")
|
||||
list := &ListResponse{Items: []*ResourceWrapper{{Value: data}}}
|
||||
index := newTestIndex(t, 1)
|
||||
|
||||
err := index.writeBatch(testContext, list)
|
||||
require.NoError(t, err)
|
||||
|
||||
assertCountEquals(t, index, 1)
|
||||
require.Equal(t, 1, len(index.allTenants()))
|
||||
assertSearchCountEquals(t, index, "*", nil, nil, 1)
|
||||
}
|
||||
|
||||
func TestIndexFolder(t *testing.T) {
|
||||
data := readTestData(t, "folder-resource.json")
|
||||
list := &ListResponse{Items: []*ResourceWrapper{{Value: data}}}
|
||||
index := newTestIndex(t, 1)
|
||||
|
||||
err := index.writeBatch(testContext, list)
|
||||
require.NoError(t, err)
|
||||
|
||||
assertCountEquals(t, index, 1)
|
||||
assertSearchCountEquals(t, index, "*", nil, nil, 1)
|
||||
}
|
||||
|
||||
func TestSearchFolder(t *testing.T) {
|
||||
dashboard := readTestData(t, "dashboard-resource.json")
|
||||
folder := readTestData(t, "folder-resource.json")
|
||||
list := &ListResponse{Items: []*ResourceWrapper{{Value: dashboard}, {Value: folder}}}
|
||||
index := newTestIndex(t, 1)
|
||||
|
||||
err := index.writeBatch(testContext, list)
|
||||
require.NoError(t, err)
|
||||
|
||||
assertCountEquals(t, index, 2)
|
||||
assertSearchCountEquals(t, index, "*", []string{"folder"}, nil, 1)
|
||||
}
|
||||
|
||||
func TestSearchDashboardsAndFoldersOnly(t *testing.T) {
|
||||
dashboard := readTestData(t, "dashboard-resource.json")
|
||||
folder := readTestData(t, "folder-resource.json")
|
||||
playlist := readTestData(t, "playlist-resource.json")
|
||||
list := &ListResponse{Items: []*ResourceWrapper{{Value: dashboard}, {Value: folder}, {Value: playlist}}}
|
||||
index := newTestIndex(t, 1)
|
||||
|
||||
err := index.writeBatch(testContext, list)
|
||||
require.NoError(t, err)
|
||||
|
||||
assertCountEquals(t, index, 3)
|
||||
assertSearchCountEquals(t, index, "*", []string{"dashboard", "folder"}, nil, 2)
|
||||
}
|
||||
|
||||
func TestLookupNames(t *testing.T) {
|
||||
records := 1000
|
||||
folders, ids := simulateFolders(records)
|
||||
list := &ListResponse{Items: []*ResourceWrapper{}}
|
||||
for _, f := range folders {
|
||||
list.Items = append(list.Items, &ResourceWrapper{Value: []byte(f)})
|
||||
}
|
||||
index := newTestIndex(t, 1)
|
||||
|
||||
err := index.writeBatch(testContext, list)
|
||||
require.NoError(t, err)
|
||||
|
||||
assertCountEquals(t, index, records)
|
||||
query := ""
|
||||
chunk := ids[:100] // query for n folders by id
|
||||
for _, id := range chunk {
|
||||
query += `"` + id + `" `
|
||||
}
|
||||
assertSearchCountEquals(t, index, query, nil, nil, int64(len(chunk)))
|
||||
}
|
||||
|
||||
func TestIndexDashboardWithTags(t *testing.T) {
|
||||
dashboard := readTestData(t, "dashboard-resource.json")
|
||||
data := readTestData(t, "dashboard-tagged-resource.json")
|
||||
data2 := readTestData(t, "dashboard-tagged-resource2.json")
|
||||
list := &ListResponse{Items: []*ResourceWrapper{{Value: dashboard}, {Value: data}, {Value: data2}}}
|
||||
index := newTestIndex(t, 2)
|
||||
|
||||
err := index.writeBatch(testContext, list)
|
||||
require.NoError(t, err)
|
||||
|
||||
assertCountEquals(t, index, 3)
|
||||
assertSearchCountEquals(t, index, "*", nil, []string{"tag1"}, 2)
|
||||
assertSearchCountEquals(t, index, "*", nil, []string{"tag4"}, 1)
|
||||
assertSearchGroupCountEquals(t, index, "*", "tags", nil, 4)
|
||||
assertSearchGroupCountEquals(t, index, "*", "tags", []string{"tag4"}, 3)
|
||||
}
|
||||
|
||||
func TestSort(t *testing.T) {
|
||||
dashboard := readTestData(t, "dashboard-resource.json")
|
||||
folder := readTestData(t, "folder-resource.json")
|
||||
playlist := readTestData(t, "playlist-resource.json")
|
||||
list := &ListResponse{Items: []*ResourceWrapper{{Value: dashboard}, {Value: folder}, {Value: playlist}}}
|
||||
index := newTestIndex(t, 1)
|
||||
|
||||
err := index.writeBatch(testContext, list)
|
||||
require.NoError(t, err)
|
||||
|
||||
assertCountEquals(t, index, 3)
|
||||
|
||||
req := &SearchRequest{Query: "*", Tenant: testTenant, Limit: 4, Offset: 0, Kind: []string{"dashboard", "folder"}, SortBy: []string{"title"}}
|
||||
results, err := index.Search(testContext, req)
|
||||
require.NoError(t, err)
|
||||
|
||||
val := results.Values[0]
|
||||
assert.Equal(t, "dashboard-a", val.Spec["title"])
|
||||
|
||||
req = &SearchRequest{Query: "*", Tenant: testTenant, Limit: 4, Offset: 0, Kind: []string{"dashboard", "folder"}, SortBy: []string{"-title"}}
|
||||
results, err = index.Search(testContext, req)
|
||||
require.NoError(t, err)
|
||||
|
||||
val = results.Values[0]
|
||||
assert.NotEqual(t, "dashboard-a", val.Spec["title"])
|
||||
}
|
||||
|
||||
func TestIndexBatch(t *testing.T) {
|
||||
index := newTestIndex(t, 1000)
|
||||
|
||||
startAll := time.Now()
|
||||
ns := namespaces()
|
||||
// simulate 10 List calls
|
||||
for i := 0; i < 10; i++ {
|
||||
list := &ListResponse{Items: loadTestItems(strconv.Itoa(i), ns)}
|
||||
start := time.Now()
|
||||
_, err := index.AddToBatches(testContext, list)
|
||||
require.NoError(t, err)
|
||||
elapsed := time.Since(start)
|
||||
fmt.Println("Time elapsed:", elapsed)
|
||||
}
|
||||
|
||||
// index all batches for each shard/tenant
|
||||
err := index.IndexBatches(testContext, 1, ns)
|
||||
require.NoError(t, err)
|
||||
|
||||
elapsed := time.Since(startAll)
|
||||
fmt.Println("Total Time elapsed:", elapsed)
|
||||
|
||||
assert.Equal(t, len(ns), len(index.shards))
|
||||
assertCountEquals(t, index, 100000)
|
||||
}
|
||||
|
||||
func loadTestItems(uid string, tenants []string) []*ResourceWrapper {
|
||||
resource := `{
|
||||
"kind": "<kind>",
|
||||
"title": "test",
|
||||
"metadata": {
|
||||
"uid": "<uid>",
|
||||
"name": "test",
|
||||
"namespace": "<ns>"
|
||||
},
|
||||
"spec": {
|
||||
"title": "test",
|
||||
"description": "test",
|
||||
"interval": "5m"
|
||||
}
|
||||
}`
|
||||
|
||||
items := []*ResourceWrapper{}
|
||||
for i := 0; i < 10000; i++ {
|
||||
res := strings.Replace(resource, "<uid>", strconv.Itoa(i)+uid, 1)
|
||||
// shuffle kinds
|
||||
kind := kinds[rand.Intn(len(kinds))]
|
||||
res = strings.Replace(res, "<kind>", kind, 1)
|
||||
// shuffle namespaces
|
||||
ns := tenants[rand.Intn(len(tenants))]
|
||||
res = strings.Replace(res, "<ns>", ns, 1)
|
||||
items = append(items, &ResourceWrapper{Value: []byte(res)})
|
||||
}
|
||||
return items
|
||||
}
|
||||
|
||||
var kinds = []string{
|
||||
"playlist",
|
||||
"folder",
|
||||
}
|
||||
|
||||
// simulate many tenants ( cloud )
|
||||
func namespaces() []string {
|
||||
ns := []string{}
|
||||
for i := 0; i < 1000; i++ {
|
||||
ns = append(ns, "tenant"+strconv.Itoa(i))
|
||||
}
|
||||
return ns
|
||||
}
|
||||
|
||||
func newTestIndex(t *testing.T, batchSize int) *Index {
|
||||
tracingCfg := tracing.NewEmptyTracingConfig()
|
||||
trace, err := tracing.ProvideService(tracingCfg)
|
||||
require.NoError(t, err)
|
||||
|
||||
return &Index{
|
||||
tracer: trace,
|
||||
shards: make(map[string]*Shard),
|
||||
log: log.New("unifiedstorage.search.index"),
|
||||
opts: Opts{
|
||||
ListLimit: 5000,
|
||||
Workers: 10,
|
||||
BatchSize: batchSize,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func assertCountEquals(t *testing.T, index *Index, expected int) {
|
||||
total, err := index.Count()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, expected, total)
|
||||
}
|
||||
|
||||
func assertSearchCountEquals(t *testing.T, index *Index, search string, kind []string, filters []string, expected int64) {
|
||||
req := &SearchRequest{Query: search, Tenant: testTenant, Limit: expected + 1, Offset: 0, Kind: kind, Filters: filters}
|
||||
start := time.Now()
|
||||
results, err := index.Search(testContext, req)
|
||||
require.NoError(t, err)
|
||||
elapsed := time.Since(start)
|
||||
fmt.Println("Search time:", elapsed)
|
||||
assert.Equal(t, expected, int64(len(results.Values)))
|
||||
}
|
||||
|
||||
func assertSearchGroupCountEquals(t *testing.T, index *Index, search string, group string, filters []string, expected int64) {
|
||||
groupBy := []*GroupBy{{Name: group, Limit: 100}}
|
||||
req := &SearchRequest{Query: search, Tenant: testTenant, Limit: 1, Offset: 0, GroupBy: groupBy, Filters: filters}
|
||||
results, err := index.Search(testContext, req)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, expected, int64(len(results.Groups)))
|
||||
}
|
||||
|
||||
func readTestData(t *testing.T, name string) []byte {
|
||||
// We can ignore the gosec G304 because this is only for tests
|
||||
// nolint:gosec
|
||||
data, err := os.ReadFile("./testdata/" + name)
|
||||
require.NoError(t, err)
|
||||
return data
|
||||
}
|
||||
|
||||
func simulateFolders(size int) ([]string, []string) {
|
||||
folders := []string{}
|
||||
ids := []string{}
|
||||
for i := 0; i < size; i++ {
|
||||
id := "folder-" + strconv.Itoa(i)
|
||||
folder := `{
|
||||
"kind": "Folder",
|
||||
"title": "test",
|
||||
"metadata": {
|
||||
"uid": "` + id + `",
|
||||
"name": "folder-` + strconv.Itoa(i) + `",
|
||||
"namespace": "default"
|
||||
},
|
||||
"spec": {
|
||||
"title": "test",
|
||||
"description": "test"
|
||||
}
|
||||
}`
|
||||
folders = append(folders, folder)
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return folders, ids
|
||||
}
|
||||
@@ -5,9 +5,8 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
_ DiagnosticsServer = (*noopService)(nil)
|
||||
_ ResourceIndexServer = (*noopService)(nil)
|
||||
_ LifecycleHooks = (*noopService)(nil)
|
||||
_ DiagnosticsServer = (*noopService)(nil)
|
||||
_ LifecycleHooks = (*noopService)(nil)
|
||||
)
|
||||
|
||||
// noopService is a helper implementation to simplify tests
|
||||
@@ -34,15 +33,3 @@ func (n *noopService) IsHealthy(context.Context, *HealthCheckRequest) (*HealthCh
|
||||
func (n *noopService) Read(context.Context, *ReadRequest) (*ReadResponse, error) {
|
||||
return nil, ErrNotImplementedYet
|
||||
}
|
||||
|
||||
func (n *noopService) Search(context.Context, *SearchRequest) (*SearchResponse, error) {
|
||||
return nil, ErrNotImplementedYet
|
||||
}
|
||||
|
||||
func (n *noopService) History(context.Context, *HistoryRequest) (*HistoryResponse, error) {
|
||||
return nil, ErrNotImplementedYet
|
||||
}
|
||||
|
||||
func (n *noopService) Origin(context.Context, *OriginRequest) (*OriginResponse, error) {
|
||||
return nil, ErrNotImplementedYet
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -323,26 +323,6 @@ 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;
|
||||
// default to bleve
|
||||
string queryType = 2;
|
||||
string tenant = 3;
|
||||
// resource kind (playlists, dashboards, etc)
|
||||
repeated string kind = 4;
|
||||
// pagination support
|
||||
int64 limit = 5;
|
||||
int64 offset = 6;
|
||||
// grouping (optional)
|
||||
repeated GroupBy groupBy = 8;
|
||||
// sorting
|
||||
repeated string sortBy = 9;
|
||||
// filters
|
||||
repeated string filters = 10;
|
||||
}
|
||||
|
||||
// Search within a single resource
|
||||
message ResourceSearchRequest {
|
||||
message Sort {
|
||||
@@ -417,10 +397,10 @@ message ResourceSearchResponse {
|
||||
ResourceTable results = 3;
|
||||
|
||||
// The total hit count
|
||||
uint64 total_hits = 4;
|
||||
int64 total_hits = 4;
|
||||
|
||||
// indicates how expensive was the query with respect to bytes read
|
||||
uint64 query_cost = 5;
|
||||
double query_cost = 5;
|
||||
|
||||
// maximum score across all fields
|
||||
double max_score = 6;
|
||||
@@ -429,22 +409,6 @@ message ResourceSearchResponse {
|
||||
map<string,Facet> facet = 7;
|
||||
}
|
||||
|
||||
message GroupBy {
|
||||
string name = 1;
|
||||
int64 limit = 2;
|
||||
}
|
||||
|
||||
message Group {
|
||||
string name = 1;
|
||||
int64 count = 2;
|
||||
}
|
||||
|
||||
// This will soon be deprecated/replaced with ResourceSearchResponse
|
||||
message SearchResponse {
|
||||
repeated ResourceWrapper items = 1;
|
||||
repeated Group groups = 2;
|
||||
}
|
||||
|
||||
message HistoryRequest {
|
||||
// Starting from the requested page (other query parameters must match!)
|
||||
string next_page_token = 1;
|
||||
@@ -749,7 +713,7 @@ service ResourceStore {
|
||||
// Unlike the ResourceStore, this service can be exposed to clients directly
|
||||
// It should be implemented with efficient indexes and does not need read-after-write semantics
|
||||
service ResourceIndex {
|
||||
rpc Search(SearchRequest) returns (SearchResponse);
|
||||
rpc Search(ResourceSearchRequest) returns (ResourceSearchResponse);
|
||||
|
||||
// Show resource history (and trash)
|
||||
rpc History(HistoryRequest) returns (HistoryResponse);
|
||||
|
||||
@@ -360,7 +360,7 @@ const (
|
||||
// Unlike the ResourceStore, this service can be exposed to clients directly
|
||||
// It should be implemented with efficient indexes and does not need read-after-write semantics
|
||||
type ResourceIndexClient interface {
|
||||
Search(ctx context.Context, in *SearchRequest, opts ...grpc.CallOption) (*SearchResponse, error)
|
||||
Search(ctx context.Context, in *ResourceSearchRequest, opts ...grpc.CallOption) (*ResourceSearchResponse, error)
|
||||
// Show resource history (and trash)
|
||||
History(ctx context.Context, in *HistoryRequest, opts ...grpc.CallOption) (*HistoryResponse, error)
|
||||
// Used for efficient provisioning
|
||||
@@ -375,9 +375,9 @@ func NewResourceIndexClient(cc grpc.ClientConnInterface) ResourceIndexClient {
|
||||
return &resourceIndexClient{cc}
|
||||
}
|
||||
|
||||
func (c *resourceIndexClient) Search(ctx context.Context, in *SearchRequest, opts ...grpc.CallOption) (*SearchResponse, error) {
|
||||
func (c *resourceIndexClient) Search(ctx context.Context, in *ResourceSearchRequest, opts ...grpc.CallOption) (*ResourceSearchResponse, error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
out := new(SearchResponse)
|
||||
out := new(ResourceSearchResponse)
|
||||
err := c.cc.Invoke(ctx, ResourceIndex_Search_FullMethodName, in, out, cOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -412,7 +412,7 @@ func (c *resourceIndexClient) Origin(ctx context.Context, in *OriginRequest, opt
|
||||
// Unlike the ResourceStore, this service can be exposed to clients directly
|
||||
// It should be implemented with efficient indexes and does not need read-after-write semantics
|
||||
type ResourceIndexServer interface {
|
||||
Search(context.Context, *SearchRequest) (*SearchResponse, error)
|
||||
Search(context.Context, *ResourceSearchRequest) (*ResourceSearchResponse, error)
|
||||
// Show resource history (and trash)
|
||||
History(context.Context, *HistoryRequest) (*HistoryResponse, error)
|
||||
// Used for efficient provisioning
|
||||
@@ -423,7 +423,7 @@ type ResourceIndexServer interface {
|
||||
type UnimplementedResourceIndexServer struct {
|
||||
}
|
||||
|
||||
func (UnimplementedResourceIndexServer) Search(context.Context, *SearchRequest) (*SearchResponse, error) {
|
||||
func (UnimplementedResourceIndexServer) Search(context.Context, *ResourceSearchRequest) (*ResourceSearchResponse, error) {
|
||||
return nil, status.Errorf(codes.Unimplemented, "method Search not implemented")
|
||||
}
|
||||
func (UnimplementedResourceIndexServer) History(context.Context, *HistoryRequest) (*HistoryResponse, error) {
|
||||
@@ -445,7 +445,7 @@ func RegisterResourceIndexServer(s grpc.ServiceRegistrar, srv ResourceIndexServe
|
||||
}
|
||||
|
||||
func _ResourceIndex_Search_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
|
||||
in := new(SearchRequest)
|
||||
in := new(ResourceSearchRequest)
|
||||
if err := dec(in); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -457,7 +457,7 @@ func _ResourceIndex_Search_Handler(srv interface{}, ctx context.Context, dec fun
|
||||
FullMethod: ResourceIndex_Search_FullMethodName,
|
||||
}
|
||||
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
|
||||
return srv.(ResourceIndexServer).Search(ctx, req.(*SearchRequest))
|
||||
return srv.(ResourceIndexServer).Search(ctx, req.(*ResourceSearchRequest))
|
||||
}
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
@@ -78,11 +78,16 @@ type searchSupport struct {
|
||||
log *slog.Logger
|
||||
storage StorageBackend
|
||||
search SearchBackend
|
||||
access authz.AccessClient
|
||||
builders *builderCache
|
||||
initWorkers int
|
||||
}
|
||||
|
||||
func newSearchSupport(opts SearchOptions, storage StorageBackend, blob BlobSupport, tracer trace.Tracer) (support *searchSupport, err error) {
|
||||
var (
|
||||
_ ResourceIndexServer = (*searchSupport)(nil)
|
||||
)
|
||||
|
||||
func newSearchSupport(opts SearchOptions, storage StorageBackend, access authz.AccessClient, blob BlobSupport, tracer trace.Tracer) (support *searchSupport, err error) {
|
||||
// No backend search support
|
||||
if opts.Backend == nil {
|
||||
return nil, nil
|
||||
@@ -93,6 +98,7 @@ func newSearchSupport(opts SearchOptions, storage StorageBackend, blob BlobSuppo
|
||||
}
|
||||
|
||||
support = &searchSupport{
|
||||
access: access,
|
||||
tracer: tracer,
|
||||
storage: storage,
|
||||
search: opts.Backend,
|
||||
@@ -113,6 +119,46 @@ func newSearchSupport(opts SearchOptions, storage StorageBackend, blob BlobSuppo
|
||||
return support, err
|
||||
}
|
||||
|
||||
// History implements ResourceIndexServer.
|
||||
func (s *searchSupport) History(context.Context, *HistoryRequest) (*HistoryResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented yet... likely should not be the serarch server")
|
||||
}
|
||||
|
||||
// Origin implements ResourceIndexServer.
|
||||
func (s *searchSupport) Origin(context.Context, *OriginRequest) (*OriginResponse, error) {
|
||||
return nil, fmt.Errorf("TBD.. rename to repository")
|
||||
}
|
||||
|
||||
// Search implements ResourceIndexServer.
|
||||
func (s *searchSupport) Search(ctx context.Context, req *ResourceSearchRequest) (*ResourceSearchResponse, error) {
|
||||
nsr := NamespacedResource{
|
||||
Group: req.Options.Key.Group,
|
||||
Namespace: req.Options.Key.Namespace,
|
||||
Resource: req.Options.Key.Resource,
|
||||
}
|
||||
idx, err := s.getOrCreateIndex(ctx, nsr)
|
||||
if err != nil {
|
||||
return &ResourceSearchResponse{
|
||||
Error: AsErrorResult(err),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Get the federated indexes
|
||||
federate := make([]ResourceIndex, len(req.Federated))
|
||||
for i, f := range req.Federated {
|
||||
nsr.Group = f.Group
|
||||
nsr.Resource = f.Resource
|
||||
federate[i], err = s.getOrCreateIndex(ctx, nsr)
|
||||
if err != nil {
|
||||
return &ResourceSearchResponse{
|
||||
Error: AsErrorResult(err),
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
return idx.Search(ctx, s.access, req, federate)
|
||||
}
|
||||
|
||||
// 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")
|
||||
@@ -160,11 +206,78 @@ func (s *searchSupport) init(ctx context.Context) error {
|
||||
}
|
||||
span.AddEvent("namespaces indexed", trace.WithAttributes(attribute.Int("namespaced_indexed", totalBatchesIndexed)))
|
||||
|
||||
s.log.Debug("TODO, listen to all events")
|
||||
// Now start listening for new events
|
||||
watchctx := context.Background() // new context?
|
||||
events, err := s.storage.WatchWriteEvents(watchctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
for {
|
||||
v := <-events
|
||||
|
||||
s.handleEvent(watchctx, v)
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Async event
|
||||
func (s *searchSupport) handleEvent(ctx context.Context, evt *WrittenEvent) {
|
||||
nsr := NamespacedResource{
|
||||
Namespace: evt.Key.Namespace,
|
||||
Group: evt.Key.Group,
|
||||
Resource: evt.Key.Resource,
|
||||
}
|
||||
|
||||
index, err := s.getOrCreateIndex(ctx, nsr)
|
||||
if err != nil {
|
||||
s.log.Warn("error getting index for watch event", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
builder, err := s.builders.get(ctx, nsr)
|
||||
if err != nil {
|
||||
s.log.Warn("error getting builder for watch event", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
doc, err := builder.BuildDocument(ctx, evt.Key, evt.ResourceVersion, evt.Value)
|
||||
if err != nil {
|
||||
s.log.Warn("error building document watch event", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
err = index.Write(doc)
|
||||
if err != nil {
|
||||
s.log.Warn("error writing document watch event", "error", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (s *searchSupport) getOrCreateIndex(ctx context.Context, key NamespacedResource) (ResourceIndex, error) {
|
||||
// TODO???
|
||||
// We want to block while building the index and return the same index for the key
|
||||
// simple mutex not great... we don't want to block while anything in building, just the same key
|
||||
|
||||
idx, err := s.search.GetIndex(ctx, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if idx == nil {
|
||||
idx, _, err = s.build(ctx, key, 10, 0) // unknown size and RV
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if idx == nil {
|
||||
return nil, fmt.Errorf("nil index after build")
|
||||
}
|
||||
}
|
||||
return idx, nil
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
@@ -145,9 +145,6 @@ type ResourceServerOptions struct {
|
||||
// The blob configuration
|
||||
Blob BlobConfig
|
||||
|
||||
// Requests based on a search index
|
||||
Index ResourceIndexServer
|
||||
|
||||
// Search options
|
||||
Search SearchOptions
|
||||
|
||||
@@ -229,7 +226,6 @@ func NewResourceServer(opts ResourceServerOptions) (ResourceServer, error) {
|
||||
tracer: opts.Tracer,
|
||||
log: logger,
|
||||
backend: opts.Backend,
|
||||
index: opts.Index,
|
||||
blob: blobstore,
|
||||
diagnostics: opts.Diagnostics,
|
||||
access: opts.AccessClient,
|
||||
@@ -242,7 +238,7 @@ func NewResourceServer(opts ResourceServerOptions) (ResourceServer, error) {
|
||||
|
||||
if opts.Search.Resources != nil {
|
||||
var err error
|
||||
s.search, err = newSearchSupport(opts.Search, s.backend, s.blob, opts.Tracer)
|
||||
s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.Tracer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -259,7 +255,6 @@ type server struct {
|
||||
backend StorageBackend
|
||||
blob BlobSupport
|
||||
search *searchSupport
|
||||
index ResourceIndexServer
|
||||
diagnostics DiagnosticsServer
|
||||
access authz.AccessClient
|
||||
writeHooks WriteAccessHooks
|
||||
@@ -923,14 +918,14 @@ func (s *server) Watch(req *WatchRequest, srv ResourceStore_WatchServer) error {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *server) Search(ctx context.Context, req *SearchRequest) (*SearchResponse, error) {
|
||||
func (s *server) Search(ctx context.Context, req *ResourceSearchRequest) (*ResourceSearchResponse, error) {
|
||||
if err := s.Init(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.index == nil {
|
||||
if s.search == nil {
|
||||
return nil, fmt.Errorf("search index not configured")
|
||||
}
|
||||
return s.index.Search(ctx, req)
|
||||
return s.search.Search(ctx, req)
|
||||
}
|
||||
|
||||
// History implements ResourceServer.
|
||||
@@ -938,7 +933,7 @@ func (s *server) History(ctx context.Context, req *HistoryRequest) (*HistoryResp
|
||||
if err := s.Init(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.index.History(ctx, req)
|
||||
return s.search.History(ctx, req)
|
||||
}
|
||||
|
||||
// Origin implements ResourceServer.
|
||||
@@ -946,33 +941,7 @@ func (s *server) Origin(ctx context.Context, req *OriginRequest) (*OriginRespons
|
||||
if err := s.Init(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.index.Origin(ctx, req)
|
||||
}
|
||||
|
||||
// Index returns the search index. If the index is not initialized, it will be initialized.
|
||||
func (s *server) Index(ctx context.Context) (*Index, error) {
|
||||
if err := s.Init(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
index := s.index.(*IndexServer)
|
||||
if index.index == nil {
|
||||
err := index.Init(ctx, s)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = index.Load(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = index.Watch(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return index.index, nil
|
||||
return s.search.Origin(ctx, req)
|
||||
}
|
||||
|
||||
// IsHealthy implements ResourceServer.
|
||||
|
||||
@@ -24,7 +24,7 @@ const tracingPrexfixBleve = "unified_search.bleve."
|
||||
var _ resource.SearchBackend = &bleveBackend{}
|
||||
var _ resource.ResourceIndex = &bleveIndex{}
|
||||
|
||||
type bleveOptions struct {
|
||||
type BleveOptions struct {
|
||||
// The root folder where file objects are saved
|
||||
Root string
|
||||
|
||||
@@ -39,14 +39,14 @@ type bleveOptions struct {
|
||||
type bleveBackend struct {
|
||||
tracer trace.Tracer
|
||||
log *slog.Logger
|
||||
opts bleveOptions
|
||||
opts BleveOptions
|
||||
|
||||
// cache info
|
||||
cache map[resource.NamespacedResource]*bleveIndex
|
||||
cacheMu sync.RWMutex
|
||||
}
|
||||
|
||||
func NewBleveBackend(opts bleveOptions, tracer trace.Tracer, reg prometheus.Registerer) *bleveBackend {
|
||||
func NewBleveBackend(opts BleveOptions, tracer trace.Tracer, reg prometheus.Registerer) *bleveBackend {
|
||||
b := &bleveBackend{
|
||||
log: slog.Default().With("logger", "bleve-backend"),
|
||||
tracer: tracer,
|
||||
@@ -247,8 +247,8 @@ func (b *bleveIndex) Search(
|
||||
return nil, err
|
||||
}
|
||||
|
||||
response.TotalHits = res.Total
|
||||
response.QueryCost = res.Cost
|
||||
response.TotalHits = int64(res.Total)
|
||||
response.QueryCost = float64(res.Cost)
|
||||
response.MaxScore = res.MaxScore
|
||||
|
||||
response.Results, err = b.hitsToTable(searchrequest.Fields, res.Hits, req.Explain)
|
||||
@@ -368,7 +368,7 @@ func toBleveSearchRequest(req *resource.ResourceSearchRequest, access authz.Acce
|
||||
// See: https://github.com/grafana/grafana/blob/v11.3.0/pkg/services/searchV2/bluge.go
|
||||
// NOTE, we likely want to pass in the already called checker because the resource server
|
||||
// will first need to check if we can see anything (or everything!) for this resource
|
||||
fmt.Printf("TODO... check authorization")
|
||||
fmt.Printf("TODO... check authorization\n")
|
||||
}
|
||||
|
||||
switch len(queries) {
|
||||
|
||||
@@ -31,7 +31,7 @@ func TestBleveBackend(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
backend := NewBleveBackend(
|
||||
bleveOptions{
|
||||
BleveOptions{
|
||||
Root: tmpdir.Name(),
|
||||
FileThreshold: 5, // with more than 5 items we create a file on disk
|
||||
},
|
||||
|
||||
@@ -3,31 +3,52 @@ package search
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/grafana/authlib/claims"
|
||||
"github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/services/store/kind/dashboard"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
)
|
||||
|
||||
// The default list of open source document builders
|
||||
type StandardDocumentBuilders struct{}
|
||||
type StandardDocumentBuilders struct {
|
||||
sql db.DB
|
||||
}
|
||||
|
||||
// Hooked up so wire can fill in different sprinkles
|
||||
func ProvideDocumentBuilders() resource.DocumentBuilderSupplier {
|
||||
return &StandardDocumentBuilders{}
|
||||
func ProvideDocumentBuilders(sql db.DB) resource.DocumentBuilderSupplier {
|
||||
return &StandardDocumentBuilders{sql}
|
||||
}
|
||||
|
||||
func (s *StandardDocumentBuilders) GetDocumentBuilders() ([]resource.DocumentBuilderInfo, error) {
|
||||
dashboards, err := DashboardBuilder(func(ctx context.Context, namespace string, blob resource.BlobSupport) (resource.DocumentBuilder, error) {
|
||||
stats := NewDashboardStatsLookup(nil) // empty stats
|
||||
dsinfo := []*dashboard.DatasourceQueryResult{{}}
|
||||
ns, err := claims.ParseNamespace(namespace)
|
||||
if err != nil && s.sql != nil {
|
||||
rows, err := s.sql.GetSqlxSession().Query(ctx, "SELECT uid,type,name,is_default FROM data_source WHERE org_id=?", ns.OrgID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
info := &dashboard.DatasourceQueryResult{}
|
||||
err = rows.Scan(&info.UID, &info.Type, &info.Name, &info.IsDefault)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dsinfo = append(dsinfo, info)
|
||||
}
|
||||
}
|
||||
return &DashboardDocumentBuilder{
|
||||
Namespace: namespace,
|
||||
Blob: blob,
|
||||
Stats: NewDashboardStatsLookup(nil), // empty stats
|
||||
DatasourceLookup: dashboard.CreateDatasourceLookup([]*dashboard.DatasourceQueryResult{{}}),
|
||||
Stats: stats,
|
||||
DatasourceLookup: dashboard.CreateDatasourceLookup(dsinfo),
|
||||
}, nil
|
||||
})
|
||||
|
||||
return []resource.DocumentBuilderInfo{
|
||||
// The default builder
|
||||
resource.DocumentBuilderInfo{
|
||||
{
|
||||
Builder: resource.StandardDocumentBuilder(),
|
||||
},
|
||||
// Dashboard builder
|
||||
|
||||
@@ -3,18 +3,18 @@ package sql
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
"github.com/grafana/authlib/claims"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/identity"
|
||||
infraDB "github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/services/authz"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/search"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl"
|
||||
)
|
||||
|
||||
@@ -54,12 +54,18 @@ func NewResourceServer(ctx context.Context, db infraDB.DB, cfg *setting.Cfg,
|
||||
opts.Backend = store
|
||||
opts.Diagnostics = store
|
||||
opts.Lifecycle = store
|
||||
opts.Search = resource.SearchOptions{
|
||||
Resources: docs,
|
||||
}
|
||||
|
||||
// Setup the search server
|
||||
if features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageSearch) {
|
||||
opts.Index = resource.NewResourceIndexServer(cfg, tracer)
|
||||
opts.Search = resource.SearchOptions{
|
||||
Backend: search.NewBleveBackend(search.BleveOptions{
|
||||
Root: filepath.Join(cfg.DataPath, "unified-search", "bleve"),
|
||||
FileThreshold: 10, // fewer than X items will use a memory index
|
||||
BatchSize: 500, // This is the batch size for how many objects to add to the index at once
|
||||
}, tracer, reg),
|
||||
Resources: docs,
|
||||
WorkerThreads: 5, // from cfg?
|
||||
}
|
||||
}
|
||||
|
||||
rs, err := resource.NewResourceServer(opts)
|
||||
@@ -67,29 +73,5 @@ func NewResourceServer(ctx context.Context, db infraDB.DB, cfg *setting.Cfg,
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Initialize the indexer if one is configured
|
||||
if opts.Index != nil {
|
||||
// TODO: Create a proper identity for the indexer
|
||||
orgId := int64(1)
|
||||
ctx = identity.WithRequester(ctx, &identity.StaticRequester{
|
||||
Type: claims.TypeServiceAccount, // system:apiserver
|
||||
UserID: 1,
|
||||
OrgID: int64(1),
|
||||
Name: "admin",
|
||||
Login: "admin",
|
||||
OrgRole: identity.RoleAdmin,
|
||||
IsGrafanaAdmin: true,
|
||||
Permissions: map[int64]map[string][]string{
|
||||
orgId: {
|
||||
"*": {"*"}, // all resources, all scopes
|
||||
},
|
||||
},
|
||||
})
|
||||
_, err = rs.(resource.ResourceIndexer).Index(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return rs, nil
|
||||
}
|
||||
|
||||
@@ -102,8 +102,8 @@ func (s *service) start(ctx context.Context) error {
|
||||
}
|
||||
|
||||
// TODO, for standalone this will need to be started from enterprise
|
||||
// Connecting to the correct remote services
|
||||
docs := search.ProvideDocumentBuilders()
|
||||
// Connecting to the correct remote services (cloudconfig for DS info and usage stats)
|
||||
docs := search.ProvideDocumentBuilders(nil)
|
||||
|
||||
server, err := NewResourceServer(ctx, s.db, s.cfg, s.features, docs, s.tracing, s.reg, authzClient)
|
||||
if err != nil {
|
||||
|
||||
@@ -1,126 +0,0 @@
|
||||
package test
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/sql"
|
||||
"github.com/grafana/grafana/pkg/util/testutil"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.org/x/net/context"
|
||||
)
|
||||
|
||||
// addResource is a helper to create a resource in unified storage
|
||||
func addResource(t *testing.T, ctx context.Context, backend sql.Backend, resourceName string, data string) {
|
||||
ir, err := resource.NewIndexedResource([]byte(data))
|
||||
require.NoError(t, err)
|
||||
_, err = backend.WriteEvent(ctx, resource.WriteEvent{
|
||||
Type: resource.WatchEvent_ADDED,
|
||||
Value: []byte(data),
|
||||
Key: &resource.ResourceKey{
|
||||
Namespace: ir.Namespace,
|
||||
Group: ir.Group,
|
||||
Resource: resourceName,
|
||||
Name: ir.Name,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestIntegrationIndexerSearch(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
|
||||
ctx := testutil.NewTestContext(t, time.Now().Add(5*time.Second))
|
||||
cfg := setting.NewCfg()
|
||||
cfg.IndexWorkers = 1
|
||||
cfg.IndexMaxBatchSize = 100
|
||||
cfg.IndexListLimit = 100
|
||||
backend, server := newServer(t, cfg)
|
||||
|
||||
playlist1 := `{
|
||||
"kind": "Playlist",
|
||||
"apiVersion": "playlist.grafana.app/v0alpha1",
|
||||
"metadata": {
|
||||
"name": "playlist dogs",
|
||||
"namespace": "tenant1",
|
||||
"uid": "1fe028dc-81bb-4268-a3ff-20899ff0a16f",
|
||||
"resourceVersion": "1",
|
||||
"creationTimestamp": "2024-01-01T12:00:00Z"
|
||||
},
|
||||
"spec": {
|
||||
"interval": "5m",
|
||||
"title": "dogs"
|
||||
}
|
||||
}`
|
||||
playlist2 := `{
|
||||
"kind": "Playlist",
|
||||
"apiVersion": "playlist.grafana.app/v0alpha1",
|
||||
"metadata": {
|
||||
"name": "playlist cats",
|
||||
"namespace": "tenant1",
|
||||
"uid": "1fe028dc-81bb-4268-a3ff-20899ff0a16f123",
|
||||
"resourceVersion": "2",
|
||||
"creationTimestamp": "2024-01-02T12:00:00Z"
|
||||
},
|
||||
"spec": {
|
||||
"interval": "5m",
|
||||
"title": "cats"
|
||||
}
|
||||
}`
|
||||
|
||||
// add playlist1 and playlist2 to unified storage
|
||||
addResource(t, ctx, backend, "playlists", playlist1)
|
||||
addResource(t, ctx, backend, "playlists", playlist2)
|
||||
|
||||
// initialize and build the search index
|
||||
indexer, ok := server.(resource.ResourceIndexer)
|
||||
if !ok {
|
||||
t.Fatal("server does not implement ResourceIndexer")
|
||||
}
|
||||
_, err := indexer.Index(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// run search tests against the index
|
||||
t.Run("can search for all resources", func(t *testing.T) {
|
||||
res, err := server.Search(ctx, &resource.SearchRequest{
|
||||
Tenant: "tenant1",
|
||||
Query: "*",
|
||||
Limit: 10,
|
||||
Offset: 0,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, res.Items, 2)
|
||||
})
|
||||
|
||||
t.Run("can search for resources by title", func(t *testing.T) {
|
||||
res, err := server.Search(ctx, &resource.SearchRequest{
|
||||
Tenant: "tenant1",
|
||||
Query: "Spec.title:dogs",
|
||||
Limit: 10,
|
||||
Offset: 0,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, res.Items, 1)
|
||||
})
|
||||
|
||||
t.Run("can filter resources by created time", func(t *testing.T) {
|
||||
res, err := server.Search(ctx, &resource.SearchRequest{
|
||||
Tenant: "tenant1",
|
||||
Query: "CreatedAt:>=\"2024-01-02\"",
|
||||
Limit: 10,
|
||||
Offset: 0,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Len(t, res.Items, 1)
|
||||
ir := resource.IndexedResource{}
|
||||
err = json.Unmarshal(res.Items[0].Value, &ir)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "playlist cats", ir.Name)
|
||||
})
|
||||
}
|
||||
@@ -56,7 +56,6 @@ func newServer(t *testing.T, cfg *setting.Cfg) (sql.Backend, resource.ResourceSe
|
||||
Backend: ret,
|
||||
Diagnostics: ret,
|
||||
Lifecycle: ret,
|
||||
Index: resource.NewResourceIndexServer(cfg, tracing.NewNoopTracerService()),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, server)
|
||||
|
||||
Reference in New Issue
Block a user