Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b2d14cc42b |
@@ -38,10 +38,10 @@ func (d *directResourceClient) GetBlob(ctx context.Context, in *resourcepb.GetBl
|
|||||||
return d.server.GetBlob(ctx, in)
|
return d.server.GetBlob(ctx, in)
|
||||||
}
|
}
|
||||||
|
|
||||||
//// GetStats implements ResourceClient.
|
// GetStats implements ResourceClient.
|
||||||
//func (d *directResourceClient) GetStats(ctx context.Context, in *resourcepb.ResourceStatsRequest, opts ...grpc.CallOption) (*resourcepb.ResourceStatsResponse, error) {
|
func (d *directResourceClient) GetStats(ctx context.Context, in *resourcepb.ResourceStatsRequest, opts ...grpc.CallOption) (*resourcepb.ResourceStatsResponse, error) {
|
||||||
// return d.server.GetStats(ctx, in)
|
return d.server.GetStats(ctx, in)
|
||||||
//}
|
}
|
||||||
|
|
||||||
// IsHealthy implements ResourceClient.
|
// IsHealthy implements ResourceClient.
|
||||||
func (d *directResourceClient) IsHealthy(ctx context.Context, in *resourcepb.HealthCheckRequest, opts ...grpc.CallOption) (*resourcepb.HealthCheckResponse, error) {
|
func (d *directResourceClient) IsHealthy(ctx context.Context, in *resourcepb.HealthCheckRequest, opts ...grpc.CallOption) (*resourcepb.HealthCheckResponse, error) {
|
||||||
@@ -53,13 +53,13 @@ func (d *directResourceClient) List(ctx context.Context, in *resourcepb.ListRequ
|
|||||||
return d.server.List(ctx, in)
|
return d.server.List(ctx, in)
|
||||||
}
|
}
|
||||||
|
|
||||||
//func (d *directResourceClient) ListManagedObjects(ctx context.Context, in *resourcepb.ListManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.ListManagedObjectsResponse, error) {
|
func (d *directResourceClient) ListManagedObjects(ctx context.Context, in *resourcepb.ListManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.ListManagedObjectsResponse, error) {
|
||||||
// return d.server.ListManagedObjects(ctx, in)
|
return d.server.ListManagedObjects(ctx, in)
|
||||||
//}
|
}
|
||||||
//
|
|
||||||
//func (d *directResourceClient) CountManagedObjects(ctx context.Context, in *resourcepb.CountManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.CountManagedObjectsResponse, error) {
|
func (d *directResourceClient) CountManagedObjects(ctx context.Context, in *resourcepb.CountManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.CountManagedObjectsResponse, error) {
|
||||||
// return d.server.CountManagedObjects(ctx, in)
|
return d.server.CountManagedObjects(ctx, in)
|
||||||
//}
|
}
|
||||||
|
|
||||||
// PutBlob implements ResourceClient.
|
// PutBlob implements ResourceClient.
|
||||||
func (d *directResourceClient) PutBlob(ctx context.Context, in *resourcepb.PutBlobRequest, opts ...grpc.CallOption) (*resourcepb.PutBlobResponse, error) {
|
func (d *directResourceClient) PutBlob(ctx context.Context, in *resourcepb.PutBlobRequest, opts ...grpc.CallOption) (*resourcepb.PutBlobResponse, error) {
|
||||||
@@ -72,9 +72,9 @@ func (d *directResourceClient) Read(ctx context.Context, in *resourcepb.ReadRequ
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Search implements ResourceClient.
|
// Search implements ResourceClient.
|
||||||
//func (d *directResourceClient) Search(ctx context.Context, in *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) {
|
func (d *directResourceClient) Search(ctx context.Context, in *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) {
|
||||||
// return d.server.Search(ctx, in)
|
return d.server.Search(ctx, in)
|
||||||
//}
|
}
|
||||||
|
|
||||||
// Update implements ResourceClient.
|
// Update implements ResourceClient.
|
||||||
func (d *directResourceClient) Update(ctx context.Context, in *resourcepb.UpdateRequest, opts ...grpc.CallOption) (*resourcepb.UpdateResponse, error) {
|
func (d *directResourceClient) Update(ctx context.Context, in *resourcepb.UpdateRequest, opts ...grpc.CallOption) (*resourcepb.UpdateResponse, error) {
|
||||||
|
|||||||
@@ -11,7 +11,6 @@ import (
|
|||||||
"k8s.io/apiserver/pkg/registry/rest"
|
"k8s.io/apiserver/pkg/registry/rest"
|
||||||
|
|
||||||
"github.com/grafana/authlib/types"
|
"github.com/grafana/authlib/types"
|
||||||
|
|
||||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||||
grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic"
|
grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic"
|
||||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||||
|
|||||||
@@ -41,6 +41,7 @@ import (
|
|||||||
"github.com/grafana/grafana/pkg/services/folder"
|
"github.com/grafana/grafana/pkg/services/folder"
|
||||||
"github.com/grafana/grafana/pkg/setting"
|
"github.com/grafana/grafana/pkg/setting"
|
||||||
"github.com/grafana/grafana/pkg/storage/unified/apistore"
|
"github.com/grafana/grafana/pkg/storage/unified/apistore"
|
||||||
|
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -77,7 +78,7 @@ func RegisterAPIService(cfg *setting.Cfg,
|
|||||||
acService accesscontrol.Service,
|
acService accesscontrol.Service,
|
||||||
accessClient authlib.AccessClient,
|
accessClient authlib.AccessClient,
|
||||||
registerer prometheus.Registerer,
|
registerer prometheus.Registerer,
|
||||||
unified resourcepb.ResourceIndexClient,
|
unified resource.ResourceClient,
|
||||||
zanzanaClient zanzana.Client,
|
zanzanaClient zanzana.Client,
|
||||||
) *FolderAPIBuilder {
|
) *FolderAPIBuilder {
|
||||||
builder := &FolderAPIBuilder{
|
builder := &FolderAPIBuilder{
|
||||||
@@ -96,7 +97,7 @@ func RegisterAPIService(cfg *setting.Cfg,
|
|||||||
return builder
|
return builder
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewAPIService(ac authlib.AccessClient, searcher resourcepb.ResourceIndexClient, features featuremgmt.FeatureToggles, zanzanaClient zanzana.Client, resourcePermissionsSvc *dynamic.NamespaceableResourceInterface) *FolderAPIBuilder {
|
func NewAPIService(ac authlib.AccessClient, searcher resource.ResourceClient, features featuremgmt.FeatureToggles, zanzanaClient zanzana.Client, resourcePermissionsSvc *dynamic.NamespaceableResourceInterface) *FolderAPIBuilder {
|
||||||
return &FolderAPIBuilder{
|
return &FolderAPIBuilder{
|
||||||
features: features,
|
features: features,
|
||||||
accessClient: ac,
|
accessClient: ac,
|
||||||
|
|||||||
@@ -615,7 +615,7 @@ func NewLocalStore(resourceInfo utils.ResourceInfo, scheme *runtime.Scheme, defa
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
client := resource.NewLocalResourceClient(server, nil)
|
client := resource.NewLocalResourceClient(server)
|
||||||
optsGetter := apistore.NewRESTOptionsGetterForClient(client, nil, defaultOpts.StorageConfig.Config, nil)
|
optsGetter := apistore.NewRESTOptionsGetterForClient(client, nil, defaultOpts.StorageConfig.Config, nil)
|
||||||
|
|
||||||
store, err := grafanaregistry.NewRegistryStore(scheme, resourceInfo, optsGetter)
|
store, err := grafanaregistry.NewRegistryStore(scheme, resourceInfo, optsGetter)
|
||||||
|
|||||||
@@ -13,6 +13,9 @@ import (
|
|||||||
"github.com/grafana/dskit/kv"
|
"github.com/grafana/dskit/kv"
|
||||||
"github.com/grafana/dskit/ring"
|
"github.com/grafana/dskit/ring"
|
||||||
ringclient "github.com/grafana/dskit/ring/client"
|
ringclient "github.com/grafana/dskit/ring/client"
|
||||||
|
"github.com/grafana/grafana/pkg/services/apiserver/options"
|
||||||
|
"github.com/grafana/grafana/pkg/storage/unified"
|
||||||
|
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||||
"github.com/prometheus/client_golang/prometheus"
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
"github.com/urfave/cli/v2"
|
"github.com/urfave/cli/v2"
|
||||||
|
|
||||||
@@ -76,6 +79,22 @@ func newModuleServer(opts Options,
|
|||||||
) (*ModuleServer, error) {
|
) (*ModuleServer, error) {
|
||||||
rootCtx, shutdownFn := context.WithCancel(context.Background())
|
rootCtx, shutdownFn := context.WithCancel(context.Background())
|
||||||
|
|
||||||
|
// TODO should inject this with Wire
|
||||||
|
apiserverCfg := cfg.SectionWithEnvOverrides("grafana-apiserver")
|
||||||
|
searchServerAddress := apiserverCfg.Key("search_server_address").MustString("")
|
||||||
|
var searchClient resourcepb.ResourceIndexClient
|
||||||
|
var err error
|
||||||
|
if searchServerAddress != "" {
|
||||||
|
storageOptions := options.StorageOptions{
|
||||||
|
SearchServerAddress: searchServerAddress,
|
||||||
|
}
|
||||||
|
searchClient, err = unified.NewSearchClient(storageOptions, features)
|
||||||
|
if err != nil {
|
||||||
|
shutdownFn()
|
||||||
|
return nil, fmt.Errorf("failed to create search client: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
s := &ModuleServer{
|
s := &ModuleServer{
|
||||||
opts: opts,
|
opts: opts,
|
||||||
apiOpts: apiOpts,
|
apiOpts: apiOpts,
|
||||||
@@ -96,6 +115,7 @@ func newModuleServer(opts Options,
|
|||||||
license: license,
|
license: license,
|
||||||
moduleRegisterer: moduleRegisterer,
|
moduleRegisterer: moduleRegisterer,
|
||||||
storageBackend: storageBackend,
|
storageBackend: storageBackend,
|
||||||
|
searchClient: searchClient,
|
||||||
hooksService: hooksService,
|
hooksService: hooksService,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -119,6 +139,7 @@ type ModuleServer struct {
|
|||||||
isInitialized bool
|
isInitialized bool
|
||||||
mtx sync.Mutex
|
mtx sync.Mutex
|
||||||
storageBackend resource.StorageBackend
|
storageBackend resource.StorageBackend
|
||||||
|
searchClient resourcepb.ResourceIndexClient
|
||||||
storageMetrics *resource.StorageMetrics
|
storageMetrics *resource.StorageMetrics
|
||||||
indexMetrics *resource.BleveIndexMetrics
|
indexMetrics *resource.BleveIndexMetrics
|
||||||
license licensing.Licensing
|
license licensing.Licensing
|
||||||
@@ -202,7 +223,7 @@ func (s *ModuleServer) Run() error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, s.registerer, docBuilders, s.storageMetrics, s.indexMetrics, s.searchServerRing, s.MemberlistKVConfig, s.httpServerRouter, s.storageBackend)
|
return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, s.registerer, docBuilders, s.storageMetrics, s.indexMetrics, s.searchServerRing, s.MemberlistKVConfig, s.httpServerRouter, s.storageBackend, s.searchClient)
|
||||||
})
|
})
|
||||||
|
|
||||||
m.RegisterModule(modules.ZanzanaServer, func() (services.Service, error) {
|
m.RegisterModule(modules.ZanzanaServer, func() (services.Service, error) {
|
||||||
|
|||||||
@@ -79,7 +79,7 @@ func NewRESTOptionsGetterMemory(originalStorageConfig storagebackend.Config, sec
|
|||||||
}
|
}
|
||||||
|
|
||||||
return NewRESTOptionsGetterForClient(
|
return NewRESTOptionsGetterForClient(
|
||||||
resource.NewLocalResourceClient(server, nil),
|
resource.NewLocalResourceClient(server),
|
||||||
secrets,
|
secrets,
|
||||||
originalStorageConfig,
|
originalStorageConfig,
|
||||||
nil,
|
nil,
|
||||||
@@ -118,7 +118,7 @@ func NewRESTOptionsGetterForFileXX(path string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
return NewRESTOptionsGetterForClient(
|
return NewRESTOptionsGetterForClient(
|
||||||
resource.NewLocalResourceClient(server, nil),
|
resource.NewLocalResourceClient(server),
|
||||||
nil, // secrets
|
nil, // secrets
|
||||||
originalStorageConfig,
|
originalStorageConfig,
|
||||||
nil,
|
nil,
|
||||||
|
|||||||
@@ -156,7 +156,7 @@ func testSetup(t testing.TB, opts ...setupOption) (context.Context, storage.Inte
|
|||||||
default:
|
default:
|
||||||
t.Fatalf("unsupported storage type: %s", setupOpts.storageType)
|
t.Fatalf("unsupported storage type: %s", setupOpts.storageType)
|
||||||
}
|
}
|
||||||
client := resource.NewLocalResourceClient(server, nil)
|
client := resource.NewLocalResourceClient(server)
|
||||||
|
|
||||||
config := storagebackend.NewDefaultConfig(setupOpts.prefix, setupOpts.codec)
|
config := storagebackend.NewDefaultConfig(setupOpts.prefix, setupOpts.codec)
|
||||||
store, destroyFunc, err := apistore.NewStorage(
|
store, destroyFunc, err := apistore.NewStorage(
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
badger "github.com/dgraph-io/badger/v4"
|
badger "github.com/dgraph-io/badger/v4"
|
||||||
|
"github.com/fullstorydev/grpchan"
|
||||||
|
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||||
otgrpc "github.com/opentracing-contrib/go-grpc"
|
otgrpc "github.com/opentracing-contrib/go-grpc"
|
||||||
"github.com/opentracing/opentracing-go"
|
"github.com/opentracing/opentracing-go"
|
||||||
"github.com/prometheus/client_golang/prometheus"
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
@@ -21,7 +23,6 @@ import (
|
|||||||
"github.com/grafana/dskit/grpcclient"
|
"github.com/grafana/dskit/grpcclient"
|
||||||
"github.com/grafana/dskit/middleware"
|
"github.com/grafana/dskit/middleware"
|
||||||
"github.com/grafana/dskit/services"
|
"github.com/grafana/dskit/services"
|
||||||
|
|
||||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||||
infraDB "github.com/grafana/grafana/pkg/infra/db"
|
infraDB "github.com/grafana/grafana/pkg/infra/db"
|
||||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||||
@@ -32,6 +33,7 @@ import (
|
|||||||
"github.com/grafana/grafana/pkg/storage/legacysql"
|
"github.com/grafana/grafana/pkg/storage/legacysql"
|
||||||
"github.com/grafana/grafana/pkg/storage/unified/federated"
|
"github.com/grafana/grafana/pkg/storage/unified/federated"
|
||||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||||
|
grpcUtils "github.com/grafana/grafana/pkg/storage/unified/resource/grpc"
|
||||||
"github.com/grafana/grafana/pkg/storage/unified/search"
|
"github.com/grafana/grafana/pkg/storage/unified/search"
|
||||||
"github.com/grafana/grafana/pkg/storage/unified/sql"
|
"github.com/grafana/grafana/pkg/storage/unified/sql"
|
||||||
"github.com/grafana/grafana/pkg/util/scheduler"
|
"github.com/grafana/grafana/pkg/util/scheduler"
|
||||||
@@ -92,6 +94,27 @@ func ProvideUnifiedStorageClient(opts *Options,
|
|||||||
return client, err
|
return client, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TODO use wire to provide to module server
|
||||||
|
func NewSearchClient(opts options.StorageOptions, features featuremgmt.FeatureToggles) (resourcepb.ResourceIndexClient, error) {
|
||||||
|
if opts.SearchServerAddress == "" {
|
||||||
|
return nil, fmt.Errorf("expecting address for search server")
|
||||||
|
}
|
||||||
|
|
||||||
|
var (
|
||||||
|
conn grpc.ClientConnInterface
|
||||||
|
err error
|
||||||
|
metrics = newClientMetrics(prometheus.NewRegistry())
|
||||||
|
)
|
||||||
|
|
||||||
|
conn, err = newGrpcConn(opts.SearchServerAddress, metrics, features, opts.GrpcClientKeepaliveTime)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
cc := grpchan.InterceptClientConn(conn, grpcUtils.UnaryClientInterceptor, grpcUtils.StreamClientInterceptor)
|
||||||
|
return resourcepb.NewResourceIndexClient(cc), nil
|
||||||
|
}
|
||||||
|
|
||||||
func newClient(opts options.StorageOptions,
|
func newClient(opts options.StorageOptions,
|
||||||
cfg *setting.Cfg,
|
cfg *setting.Cfg,
|
||||||
features featuremgmt.FeatureToggles,
|
features featuremgmt.FeatureToggles,
|
||||||
@@ -136,7 +159,7 @@ func newClient(opts options.StorageOptions,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
return resource.NewLocalResourceClient(server, nil), nil
|
return resource.NewLocalResourceClient(server), nil
|
||||||
|
|
||||||
case options.StorageTypeUnifiedGrpc:
|
case options.StorageTypeUnifiedGrpc:
|
||||||
if opts.Address == "" {
|
if opts.Address == "" {
|
||||||
@@ -225,11 +248,11 @@ func newClient(opts options.StorageOptions,
|
|||||||
serverOptions.OverridesService = overridesSvc
|
serverOptions.OverridesService = overridesSvc
|
||||||
}
|
}
|
||||||
|
|
||||||
server, searchServer, err := sql.NewResourceServer(serverOptions)
|
server, err := sql.NewResourceServer(serverOptions)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
return resource.NewLocalResourceClient(server, searchServer), nil
|
return resource.NewLocalResourceClient(server), nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -31,14 +31,11 @@ import (
|
|||||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||||
)
|
)
|
||||||
|
|
||||||
type SearchClient interface {
|
|
||||||
resourcepb.ResourceIndexClient
|
|
||||||
resourcepb.ManagedObjectIndexClient
|
|
||||||
}
|
|
||||||
|
|
||||||
//go:generate mockery --name ResourceClient --structname MockResourceClient --inpackage --filename client_mock.go --with-expecter
|
//go:generate mockery --name ResourceClient --structname MockResourceClient --inpackage --filename client_mock.go --with-expecter
|
||||||
type ResourceClient interface {
|
type ResourceClient interface {
|
||||||
resourcepb.ResourceStoreClient
|
resourcepb.ResourceStoreClient
|
||||||
|
resourcepb.ResourceIndexClient
|
||||||
|
resourcepb.ManagedObjectIndexClient
|
||||||
resourcepb.BulkStoreClient
|
resourcepb.BulkStoreClient
|
||||||
resourcepb.BlobStoreClient
|
resourcepb.BlobStoreClient
|
||||||
resourcepb.DiagnosticsClient
|
resourcepb.DiagnosticsClient
|
||||||
@@ -103,6 +100,8 @@ func NewLocalResourceClient(server ResourceServer) ResourceClient {
|
|||||||
grpcAuthInt := grpcutils.NewUnsafeAuthenticator(tracer)
|
grpcAuthInt := grpcutils.NewUnsafeAuthenticator(tracer)
|
||||||
for _, desc := range []*grpc.ServiceDesc{
|
for _, desc := range []*grpc.ServiceDesc{
|
||||||
&resourcepb.ResourceStore_ServiceDesc,
|
&resourcepb.ResourceStore_ServiceDesc,
|
||||||
|
&resourcepb.ResourceIndex_ServiceDesc,
|
||||||
|
&resourcepb.ManagedObjectIndex_ServiceDesc,
|
||||||
&resourcepb.BlobStore_ServiceDesc,
|
&resourcepb.BlobStore_ServiceDesc,
|
||||||
&resourcepb.BulkStore_ServiceDesc,
|
&resourcepb.BulkStore_ServiceDesc,
|
||||||
&resourcepb.Diagnostics_ServiceDesc,
|
&resourcepb.Diagnostics_ServiceDesc,
|
||||||
|
|||||||
@@ -127,10 +127,7 @@ type SearchBackend interface {
|
|||||||
GetOpenIndexes() []NamespacedResource
|
GetOpenIndexes() []NamespacedResource
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ SearchServer = &searchSupport{}
|
// This supports indexing+search regardless of implementation
|
||||||
|
|
||||||
// This supports indexing+search regardless of implementation.
|
|
||||||
// Implements SearchServer interface.
|
|
||||||
type searchSupport struct {
|
type searchSupport struct {
|
||||||
log log.Logger
|
log log.Logger
|
||||||
storage StorageBackend
|
storage StorageBackend
|
||||||
@@ -163,10 +160,6 @@ var (
|
|||||||
_ resourcepb.ManagedObjectIndexServer = (*searchSupport)(nil)
|
_ resourcepb.ManagedObjectIndexServer = (*searchSupport)(nil)
|
||||||
)
|
)
|
||||||
|
|
||||||
func NewSearchServer(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, indexMetrics *BleveIndexMetrics, ownsIndexFn func(key NamespacedResource) (bool, error)) (SearchServer, error) {
|
|
||||||
return newSearchSupport(opts, storage, access, blob, indexMetrics, ownsIndexFn)
|
|
||||||
}
|
|
||||||
|
|
||||||
func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, indexMetrics *BleveIndexMetrics, ownsIndexFn func(key NamespacedResource) (bool, error)) (support *searchSupport, err error) {
|
func newSearchSupport(opts SearchOptions, storage StorageBackend, access types.AccessClient, blob BlobSupport, indexMetrics *BleveIndexMetrics, ownsIndexFn func(key NamespacedResource) (bool, error)) (support *searchSupport, err error) {
|
||||||
// No backend search support
|
// No backend search support
|
||||||
if opts.Backend == nil {
|
if opts.Backend == nil {
|
||||||
@@ -605,15 +598,6 @@ func (s *searchSupport) buildIndexes(ctx context.Context) (int, error) {
|
|||||||
return totalBatchesIndexed, nil
|
return totalBatchesIndexed, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *searchSupport) Init(ctx context.Context) error {
|
|
||||||
return s.init(ctx)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *searchSupport) Stop(_ context.Context) error {
|
|
||||||
s.stop()
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *searchSupport) init(ctx context.Context) error {
|
func (s *searchSupport) init(ctx context.Context) error {
|
||||||
origCtx := ctx
|
origCtx := ctx
|
||||||
|
|
||||||
|
|||||||
@@ -60,7 +60,7 @@ func ProvideSearchDistributorServer(cfg *setting.Cfg, features featuremgmt.Featu
|
|||||||
}
|
}
|
||||||
|
|
||||||
type RingClient struct {
|
type RingClient struct {
|
||||||
Client SearchClient
|
Client ResourceClient
|
||||||
grpc_health_v1.HealthClient
|
grpc_health_v1.HealthClient
|
||||||
Conn *grpc.ClientConn
|
Conn *grpc.ClientConn
|
||||||
}
|
}
|
||||||
@@ -99,7 +99,7 @@ var (
|
|||||||
func (ds *distributorServer) Search(ctx context.Context, r *resourcepb.ResourceSearchRequest) (*resourcepb.ResourceSearchResponse, error) {
|
func (ds *distributorServer) Search(ctx context.Context, r *resourcepb.ResourceSearchRequest) (*resourcepb.ResourceSearchResponse, error) {
|
||||||
ctx, span := ds.tracing.Start(ctx, "distributor.Search")
|
ctx, span := ds.tracing.Start(ctx, "distributor.Search")
|
||||||
defer span.End()
|
defer span.End()
|
||||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Options.Key.Namespace)
|
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Options.Key.Namespace, "Search")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -110,7 +110,7 @@ func (ds *distributorServer) Search(ctx context.Context, r *resourcepb.ResourceS
|
|||||||
func (ds *distributorServer) GetStats(ctx context.Context, r *resourcepb.ResourceStatsRequest) (*resourcepb.ResourceStatsResponse, error) {
|
func (ds *distributorServer) GetStats(ctx context.Context, r *resourcepb.ResourceStatsRequest) (*resourcepb.ResourceStatsResponse, error) {
|
||||||
ctx, span := ds.tracing.Start(ctx, "distributor.GetStats")
|
ctx, span := ds.tracing.Start(ctx, "distributor.GetStats")
|
||||||
defer span.End()
|
defer span.End()
|
||||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace)
|
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace, "GetStats")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -215,7 +215,7 @@ func (ds *distributorServer) RebuildIndexes(ctx context.Context, r *resourcepb.R
|
|||||||
func (ds *distributorServer) CountManagedObjects(ctx context.Context, r *resourcepb.CountManagedObjectsRequest) (*resourcepb.CountManagedObjectsResponse, error) {
|
func (ds *distributorServer) CountManagedObjects(ctx context.Context, r *resourcepb.CountManagedObjectsRequest) (*resourcepb.CountManagedObjectsResponse, error) {
|
||||||
ctx, span := ds.tracing.Start(ctx, "distributor.CountManagedObjects")
|
ctx, span := ds.tracing.Start(ctx, "distributor.CountManagedObjects")
|
||||||
defer span.End()
|
defer span.End()
|
||||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace)
|
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace, "CountManagedObjects")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -226,7 +226,7 @@ func (ds *distributorServer) CountManagedObjects(ctx context.Context, r *resourc
|
|||||||
func (ds *distributorServer) ListManagedObjects(ctx context.Context, r *resourcepb.ListManagedObjectsRequest) (*resourcepb.ListManagedObjectsResponse, error) {
|
func (ds *distributorServer) ListManagedObjects(ctx context.Context, r *resourcepb.ListManagedObjectsRequest) (*resourcepb.ListManagedObjectsResponse, error) {
|
||||||
ctx, span := ds.tracing.Start(ctx, "distributor.ListManagedObjects")
|
ctx, span := ds.tracing.Start(ctx, "distributor.ListManagedObjects")
|
||||||
defer span.End()
|
defer span.End()
|
||||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace)
|
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace, "ListManagedObjects")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -234,7 +234,7 @@ func (ds *distributorServer) ListManagedObjects(ctx context.Context, r *resource
|
|||||||
return client.ListManagedObjects(ctx, r)
|
return client.ListManagedObjects(ctx, r)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ds *distributorServer) getClientToDistributeRequest(ctx context.Context, namespace string) (context.Context, SearchClient, error) {
|
func (ds *distributorServer) getClientToDistributeRequest(ctx context.Context, namespace string, methodName string) (context.Context, ResourceClient, error) {
|
||||||
ringHasher := fnv.New32a()
|
ringHasher := fnv.New32a()
|
||||||
_, err := ringHasher.Write([]byte(namespace))
|
_, err := ringHasher.Write([]byte(namespace))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -32,17 +32,12 @@ import (
|
|||||||
|
|
||||||
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/resource")
|
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/resource")
|
||||||
|
|
||||||
type SearchServer interface {
|
|
||||||
LifecycleHooks
|
|
||||||
|
|
||||||
resourcepb.ResourceIndexServer
|
|
||||||
resourcepb.ManagedObjectIndexServer
|
|
||||||
}
|
|
||||||
|
|
||||||
// ResourceServer implements all gRPC services
|
// ResourceServer implements all gRPC services
|
||||||
type ResourceServer interface {
|
type ResourceServer interface {
|
||||||
resourcepb.ResourceStoreServer
|
resourcepb.ResourceStoreServer
|
||||||
resourcepb.BulkStoreServer
|
resourcepb.BulkStoreServer
|
||||||
|
resourcepb.ResourceIndexServer
|
||||||
|
resourcepb.ManagedObjectIndexServer
|
||||||
resourcepb.BlobStoreServer
|
resourcepb.BlobStoreServer
|
||||||
resourcepb.DiagnosticsServer
|
resourcepb.DiagnosticsServer
|
||||||
resourcepb.QuotasServer
|
resourcepb.QuotasServer
|
||||||
@@ -225,8 +220,10 @@ type ResourceServerOptions struct {
|
|||||||
Blob BlobConfig
|
Blob BlobConfig
|
||||||
|
|
||||||
// Search options
|
// Search options
|
||||||
SearchOptions SearchOptions // TODO: needed?
|
Search SearchOptions
|
||||||
Search SearchServer
|
|
||||||
|
// to be used by storage
|
||||||
|
SearchClient resourcepb.ResourceIndexClient
|
||||||
|
|
||||||
// Quota service
|
// Quota service
|
||||||
OverridesService *OverridesService
|
OverridesService *OverridesService
|
||||||
@@ -255,12 +252,16 @@ type ResourceServerOptions struct {
|
|||||||
|
|
||||||
storageMetrics *StorageMetrics
|
storageMetrics *StorageMetrics
|
||||||
|
|
||||||
|
IndexMetrics *BleveIndexMetrics
|
||||||
|
|
||||||
// MaxPageSizeBytes is the maximum size of a page in bytes.
|
// MaxPageSizeBytes is the maximum size of a page in bytes.
|
||||||
MaxPageSizeBytes int
|
MaxPageSizeBytes int
|
||||||
|
|
||||||
// QOSQueue is the quality of service queue used to enqueue
|
// QOSQueue is the quality of service queue used to enqueue
|
||||||
QOSQueue QOSEnqueuer
|
QOSQueue QOSEnqueuer
|
||||||
QOSConfig QueueConfig
|
QOSConfig QueueConfig
|
||||||
|
|
||||||
|
OwnsIndexFn func(key NamespacedResource) (bool, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewResourceServer(opts ResourceServerOptions) (*server, error) {
|
func NewResourceServer(opts ResourceServerOptions) (*server, error) {
|
||||||
@@ -343,25 +344,24 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
|
|||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
cancel: cancel,
|
cancel: cancel,
|
||||||
storageMetrics: opts.storageMetrics,
|
storageMetrics: opts.storageMetrics,
|
||||||
|
indexMetrics: opts.IndexMetrics,
|
||||||
maxPageSizeBytes: opts.MaxPageSizeBytes,
|
maxPageSizeBytes: opts.MaxPageSizeBytes,
|
||||||
reg: opts.Reg,
|
reg: opts.Reg,
|
||||||
queue: opts.QOSQueue,
|
queue: opts.QOSQueue,
|
||||||
queueConfig: opts.QOSConfig,
|
queueConfig: opts.QOSConfig,
|
||||||
overridesService: opts.OverridesService,
|
overridesService: opts.OverridesService,
|
||||||
search: opts.Search,
|
searchClient: opts.SearchClient,
|
||||||
|
|
||||||
artificialSuccessfulWriteDelay: opts.SearchOptions.IndexMinUpdateInterval,
|
artificialSuccessfulWriteDelay: opts.Search.IndexMinUpdateInterval,
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
if opts.Search.Resources != nil {
|
||||||
if opts.Search.Resources != nil {
|
var err error
|
||||||
var err error
|
s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.IndexMetrics, opts.OwnsIndexFn)
|
||||||
s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.IndexMetrics, opts.OwnsIndexFn)
|
if err != nil {
|
||||||
if err != nil {
|
return nil, err
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
*/
|
}
|
||||||
|
|
||||||
err := s.Init(ctx)
|
err := s.Init(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -379,7 +379,7 @@ type server struct {
|
|||||||
backend StorageBackend
|
backend StorageBackend
|
||||||
blob BlobSupport
|
blob BlobSupport
|
||||||
secure secrets.InlineSecureValueSupport
|
secure secrets.InlineSecureValueSupport
|
||||||
search SearchServer
|
search *searchSupport
|
||||||
diagnostics resourcepb.DiagnosticsServer
|
diagnostics resourcepb.DiagnosticsServer
|
||||||
access claims.AccessClient
|
access claims.AccessClient
|
||||||
writeHooks WriteAccessHooks
|
writeHooks WriteAccessHooks
|
||||||
@@ -390,6 +390,9 @@ type server struct {
|
|||||||
indexMetrics *BleveIndexMetrics
|
indexMetrics *BleveIndexMetrics
|
||||||
overridesService *OverridesService
|
overridesService *OverridesService
|
||||||
|
|
||||||
|
// only to be used with storage server for field selector search
|
||||||
|
searchClient resourcepb.ResourceIndexClient
|
||||||
|
|
||||||
// Background watch task -- this has permissions for everything
|
// Background watch task -- this has permissions for everything
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
@@ -426,6 +429,11 @@ func (s *server) Init(ctx context.Context) error {
|
|||||||
s.initErr = s.overridesService.init(ctx)
|
s.initErr = s.overridesService.init(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initialize the search index
|
||||||
|
if s.initErr == nil && s.search != nil {
|
||||||
|
s.initErr = s.search.init(ctx)
|
||||||
|
}
|
||||||
|
|
||||||
// Start watching for changes
|
// Start watching for changes
|
||||||
if s.initErr == nil {
|
if s.initErr == nil {
|
||||||
s.initErr = s.initWatcher()
|
s.initErr = s.initWatcher()
|
||||||
@@ -450,6 +458,10 @@ func (s *server) Stop(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if s.search != nil {
|
||||||
|
s.search.stop()
|
||||||
|
}
|
||||||
|
|
||||||
if s.overridesService != nil {
|
if s.overridesService != nil {
|
||||||
if err := s.overridesService.stop(ctx); err != nil {
|
if err := s.overridesService.stop(ctx); err != nil {
|
||||||
stopFailed = true
|
stopFailed = true
|
||||||
@@ -1054,7 +1066,8 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
|
|||||||
// TODO: What to do about RV and version_match fields?
|
// TODO: What to do about RV and version_match fields?
|
||||||
// If we get here, we're doing list with selectable fields. Let's do search instead, since
|
// If we get here, we're doing list with selectable fields. Let's do search instead, since
|
||||||
// we index all selectable fields, and fetch resulting documents one by one.
|
// we index all selectable fields, and fetch resulting documents one by one.
|
||||||
if s.search != nil && req.Source == resourcepb.ListRequest_STORE && (len(req.Options.Fields) > 0) {
|
|
||||||
|
if (s.search != nil || s.searchClient != nil) && req.Source == resourcepb.ListRequest_STORE && (len(req.Options.Fields) > 0) {
|
||||||
if req.Options.Key.Namespace == "" {
|
if req.Options.Key.Namespace == "" {
|
||||||
return &resourcepb.ListResponse{
|
return &resourcepb.ListResponse{
|
||||||
Error: NewBadRequestError("namespace must be specified for list with filter"),
|
Error: NewBadRequestError("namespace must be specified for list with filter"),
|
||||||
@@ -1070,7 +1083,13 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
|
|||||||
// Permission: 0, // Not needed, default is List
|
// Permission: 0, // Not needed, default is List
|
||||||
}
|
}
|
||||||
|
|
||||||
searchResp, err := s.search.Search(ctx, srq)
|
var searchResp *resourcepb.ResourceSearchResponse
|
||||||
|
var err error
|
||||||
|
if s.searchClient != nil {
|
||||||
|
searchResp, err = s.searchClient.Search(ctx, srq)
|
||||||
|
} else {
|
||||||
|
searchResp, err = s.search.Search(ctx, srq)
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -1427,6 +1446,47 @@ func (s *server) Watch(req *resourcepb.WatchRequest, srv resourcepb.ResourceStor
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *server) Search(ctx context.Context, req *resourcepb.ResourceSearchRequest) (*resourcepb.ResourceSearchResponse, error) {
|
||||||
|
if s.search == nil {
|
||||||
|
return nil, fmt.Errorf("search index not configured")
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.search.Search(ctx, req)
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetStats implements ResourceServer.
|
||||||
|
func (s *server) GetStats(ctx context.Context, req *resourcepb.ResourceStatsRequest) (*resourcepb.ResourceStatsResponse, error) {
|
||||||
|
if err := s.Init(ctx); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
if s.search == nil {
|
||||||
|
// If the backend implements "GetStats", we can use it
|
||||||
|
srv, ok := s.backend.(resourcepb.ResourceIndexServer)
|
||||||
|
if ok {
|
||||||
|
return srv.GetStats(ctx, req)
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("search index not configured")
|
||||||
|
}
|
||||||
|
return s.search.GetStats(ctx, req)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *server) ListManagedObjects(ctx context.Context, req *resourcepb.ListManagedObjectsRequest) (*resourcepb.ListManagedObjectsResponse, error) {
|
||||||
|
if s.search == nil {
|
||||||
|
return nil, fmt.Errorf("search index not configured")
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.search.ListManagedObjects(ctx, req)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *server) CountManagedObjects(ctx context.Context, req *resourcepb.CountManagedObjectsRequest) (*resourcepb.CountManagedObjectsResponse, error) {
|
||||||
|
if s.search == nil {
|
||||||
|
return nil, fmt.Errorf("search index not configured")
|
||||||
|
}
|
||||||
|
|
||||||
|
return s.search.CountManagedObjects(ctx, req)
|
||||||
|
}
|
||||||
|
|
||||||
// IsHealthy implements ResourceServer.
|
// IsHealthy implements ResourceServer.
|
||||||
func (s *server) IsHealthy(ctx context.Context, req *resourcepb.HealthCheckRequest) (*resourcepb.HealthCheckResponse, error) {
|
func (s *server) IsHealthy(ctx context.Context, req *resourcepb.HealthCheckRequest) (*resourcepb.HealthCheckResponse, error) {
|
||||||
return s.diagnostics.IsHealthy(ctx, req)
|
return s.diagnostics.IsHealthy(ctx, req)
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
|
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||||
"github.com/prometheus/client_golang/prometheus"
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
"go.opentelemetry.io/otel/trace"
|
"go.opentelemetry.io/otel/trace"
|
||||||
|
|
||||||
@@ -37,6 +38,7 @@ type ServerOptions struct {
|
|||||||
Tracer trace.Tracer
|
Tracer trace.Tracer
|
||||||
Reg prometheus.Registerer
|
Reg prometheus.Registerer
|
||||||
AccessClient types.AccessClient
|
AccessClient types.AccessClient
|
||||||
|
SearchClient resourcepb.ResourceIndexClient
|
||||||
SearchOptions resource.SearchOptions
|
SearchOptions resource.SearchOptions
|
||||||
StorageMetrics *resource.StorageMetrics
|
StorageMetrics *resource.StorageMetrics
|
||||||
IndexMetrics *resource.BleveIndexMetrics
|
IndexMetrics *resource.BleveIndexMetrics
|
||||||
@@ -46,7 +48,7 @@ type ServerOptions struct {
|
|||||||
OwnsIndexFn func(key resource.NamespacedResource) (bool, error)
|
OwnsIndexFn func(key resource.NamespacedResource) (bool, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.SearchServer, error) {
|
func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
|
||||||
apiserverCfg := opts.Cfg.SectionWithEnvOverrides("grafana-apiserver")
|
apiserverCfg := opts.Cfg.SectionWithEnvOverrides("grafana-apiserver")
|
||||||
|
|
||||||
if opts.SecureValues == nil && opts.Cfg != nil && opts.Cfg.SecretsManagement.GrpcClientEnable {
|
if opts.SecureValues == nil && opts.Cfg != nil && opts.Cfg.SecretsManagement.GrpcClientEnable {
|
||||||
@@ -57,7 +59,7 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.Se
|
|||||||
nil, // not needed for gRPC client mode
|
nil, // not needed for gRPC client mode
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, fmt.Errorf("failed to create inline secure value service: %w", err)
|
return nil, fmt.Errorf("failed to create inline secure value service: %w", err)
|
||||||
}
|
}
|
||||||
opts.SecureValues = inlineSecureValueService
|
opts.SecureValues = inlineSecureValueService
|
||||||
}
|
}
|
||||||
@@ -77,7 +79,7 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.Se
|
|||||||
dir := strings.Replace(serverOptions.Blob.URL, "./data", opts.Cfg.DataPath, 1)
|
dir := strings.Replace(serverOptions.Blob.URL, "./data", opts.Cfg.DataPath, 1)
|
||||||
err := os.MkdirAll(dir, 0700)
|
err := os.MkdirAll(dir, 0700)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
serverOptions.Blob.URL = "file:///" + dir
|
serverOptions.Blob.URL = "file:///" + dir
|
||||||
}
|
}
|
||||||
@@ -94,7 +96,7 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.Se
|
|||||||
} else {
|
} else {
|
||||||
eDB, err := dbimpl.ProvideResourceDB(opts.DB, opts.Cfg, opts.Tracer)
|
eDB, err := dbimpl.ProvideResourceDB(opts.DB, opts.Cfg, opts.Tracer)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
isHA := isHighAvailabilityEnabled(opts.Cfg.SectionWithEnvOverrides("database"),
|
isHA := isHighAvailabilityEnabled(opts.Cfg.SectionWithEnvOverrides("database"),
|
||||||
@@ -108,31 +110,24 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, resource.Se
|
|||||||
LastImportTimeMaxAge: opts.SearchOptions.MaxIndexAge, // No need to keep last_import_times older than max index age.
|
LastImportTimeMaxAge: opts.SearchOptions.MaxIndexAge, // No need to keep last_import_times older than max index age.
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
serverOptions.Backend = backend
|
serverOptions.Backend = backend
|
||||||
serverOptions.Diagnostics = backend
|
serverOptions.Diagnostics = backend
|
||||||
serverOptions.Lifecycle = backend
|
serverOptions.Lifecycle = backend
|
||||||
}
|
}
|
||||||
|
|
||||||
search, err := resource.NewSearchServer(opts.SearchOptions, opts.Backend, opts.AccessClient, nil, opts.IndexMetrics, opts.OwnsIndexFn)
|
// use the search client when search isnt initialized. Dont need both.
|
||||||
if err != nil {
|
if opts.SearchOptions.Resources == nil {
|
||||||
return nil, nil, fmt.Errorf("failed to initialize search: %w", err)
|
serverOptions.SearchClient = opts.SearchClient
|
||||||
}
|
}
|
||||||
|
serverOptions.Search = opts.SearchOptions
|
||||||
if err := search.Init(context.Background()); err != nil {
|
serverOptions.IndexMetrics = opts.IndexMetrics
|
||||||
return nil, nil, fmt.Errorf("failed to initialize search: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
serverOptions.Search = search
|
|
||||||
serverOptions.QOSQueue = opts.QOSQueue
|
serverOptions.QOSQueue = opts.QOSQueue
|
||||||
|
serverOptions.OwnsIndexFn = opts.OwnsIndexFn
|
||||||
serverOptions.OverridesService = opts.OverridesService
|
serverOptions.OverridesService = opts.OverridesService
|
||||||
|
|
||||||
rs, err := resource.NewResourceServer(serverOptions)
|
return resource.NewResourceServer(serverOptions)
|
||||||
if err != nil {
|
|
||||||
_ = search.Stop(context.Background())
|
|
||||||
}
|
|
||||||
return rs, nil, err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// isHighAvailabilityEnabled determines if high availability mode should
|
// isHighAvailabilityEnabled determines if high availability mode should
|
||||||
|
|||||||
@@ -59,12 +59,13 @@ type service struct {
|
|||||||
subservicesWatcher *services.FailureWatcher
|
subservicesWatcher *services.FailureWatcher
|
||||||
hasSubservices bool
|
hasSubservices bool
|
||||||
|
|
||||||
backend resource.StorageBackend
|
backend resource.StorageBackend
|
||||||
cfg *setting.Cfg
|
searchClient resourcepb.ResourceIndexClient
|
||||||
features featuremgmt.FeatureToggles
|
cfg *setting.Cfg
|
||||||
db infraDB.DB
|
features featuremgmt.FeatureToggles
|
||||||
stopCh chan struct{}
|
db infraDB.DB
|
||||||
stoppedCh chan error
|
stopCh chan struct{}
|
||||||
|
stoppedCh chan error
|
||||||
|
|
||||||
handler grpcserver.Provider
|
handler grpcserver.Provider
|
||||||
|
|
||||||
@@ -99,6 +100,7 @@ func ProvideUnifiedStorageGrpcService(
|
|||||||
memberlistKVConfig kv.Config,
|
memberlistKVConfig kv.Config,
|
||||||
httpServerRouter *mux.Router,
|
httpServerRouter *mux.Router,
|
||||||
backend resource.StorageBackend,
|
backend resource.StorageBackend,
|
||||||
|
searchClient resourcepb.ResourceIndexClient,
|
||||||
) (UnifiedStorageGrpcService, error) {
|
) (UnifiedStorageGrpcService, error) {
|
||||||
var err error
|
var err error
|
||||||
tracer := otel.Tracer("unified-storage")
|
tracer := otel.Tracer("unified-storage")
|
||||||
@@ -112,6 +114,7 @@ func ProvideUnifiedStorageGrpcService(
|
|||||||
|
|
||||||
s := &service{
|
s := &service{
|
||||||
backend: backend,
|
backend: backend,
|
||||||
|
searchClient: searchClient,
|
||||||
cfg: cfg,
|
cfg: cfg,
|
||||||
features: features,
|
features: features,
|
||||||
stopCh: make(chan struct{}),
|
stopCh: make(chan struct{}),
|
||||||
@@ -272,6 +275,7 @@ func (s *service) starting(ctx context.Context) error {
|
|||||||
Tracer: s.tracing,
|
Tracer: s.tracing,
|
||||||
Reg: s.reg,
|
Reg: s.reg,
|
||||||
AccessClient: authzClient,
|
AccessClient: authzClient,
|
||||||
|
SearchClient: s.searchClient,
|
||||||
SearchOptions: searchOptions,
|
SearchOptions: searchOptions,
|
||||||
StorageMetrics: s.storageMetrics,
|
StorageMetrics: s.storageMetrics,
|
||||||
IndexMetrics: s.indexMetrics,
|
IndexMetrics: s.indexMetrics,
|
||||||
@@ -291,7 +295,7 @@ func (s *service) starting(ctx context.Context) error {
|
|||||||
serverOptions.OverridesService = overridesSvc
|
serverOptions.OverridesService = overridesSvc
|
||||||
}
|
}
|
||||||
|
|
||||||
server, searchServer, err := NewResourceServer(serverOptions)
|
server, err := NewResourceServer(serverOptions)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -308,8 +312,8 @@ func (s *service) starting(ctx context.Context) error {
|
|||||||
srv := s.handler.GetServer()
|
srv := s.handler.GetServer()
|
||||||
resourcepb.RegisterResourceStoreServer(srv, server)
|
resourcepb.RegisterResourceStoreServer(srv, server)
|
||||||
resourcepb.RegisterBulkStoreServer(srv, server)
|
resourcepb.RegisterBulkStoreServer(srv, server)
|
||||||
resourcepb.RegisterResourceIndexServer(srv, searchServer)
|
resourcepb.RegisterResourceIndexServer(srv, server)
|
||||||
resourcepb.RegisterManagedObjectIndexServer(srv, searchServer)
|
resourcepb.RegisterManagedObjectIndexServer(srv, server)
|
||||||
resourcepb.RegisterBlobStoreServer(srv, server)
|
resourcepb.RegisterBlobStoreServer(srv, server)
|
||||||
resourcepb.RegisterDiagnosticsServer(srv, server)
|
resourcepb.RegisterDiagnosticsServer(srv, server)
|
||||||
resourcepb.RegisterQuotasServer(srv, server)
|
resourcepb.RegisterQuotasServer(srv, server)
|
||||||
|
|||||||
Reference in New Issue
Block a user