Compare commits

..
14 changed files with 190 additions and 104 deletions
+14 -14
View File
@@ -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"
+3 -2
View File
@@ -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,
+1 -1
View File
@@ -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)
+22 -1
View File
@@ -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) {
+2 -2
View File
@@ -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,
+1 -1
View File
@@ -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(
+27 -4
View File
@@ -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
} }
} }
+4 -5
View File
@@ -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,
+1 -17
View File
@@ -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 {
+82 -22
View File
@@ -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)
+14 -19
View File
@@ -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
+13 -9
View File
@@ -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)