Compare commits
14 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c8da64e4eb | |||
| 2fab497c18 | |||
| b0bb71f834 | |||
| a7aa55f908 | |||
| 0a61846b5e | |||
| 785cc739ee | |||
| 4913baaf04 | |||
| c8f1efe7c7 | |||
| b91ca14f48 | |||
| 2aedbdb76f | |||
| 0d7f46c08a | |||
| 1b52718c23 | |||
| e61e406440 | |||
| 5cb4c311dc |
@@ -4020,11 +4020,6 @@
|
||||
"count": 1
|
||||
}
|
||||
},
|
||||
"public/app/plugins/datasource/parca/webpack.config.ts": {
|
||||
"no-barrel-files/no-barrel-files": {
|
||||
"count": 1
|
||||
}
|
||||
},
|
||||
"public/app/plugins/datasource/prometheus/configuration/AzureAuthSettings.tsx": {
|
||||
"no-restricted-syntax": {
|
||||
"count": 1
|
||||
|
||||
@@ -585,6 +585,8 @@ module.exports = [
|
||||
// FIXME: Remove once all enterprise issues are fixed -
|
||||
// we don't have a suppressions file/approach for enterprise code yet
|
||||
...enterpriseIgnores,
|
||||
// Ignore decoupled plugin webpack configs
|
||||
'public/app/**/webpack.config.ts',
|
||||
],
|
||||
rules: {
|
||||
'no-barrel-files/no-barrel-files': 'error',
|
||||
|
||||
+6
@@ -246,6 +246,8 @@ const injectedRtkApi = api
|
||||
facetLimit: queryArg.facetLimit,
|
||||
tags: queryArg.tags,
|
||||
libraryPanel: queryArg.libraryPanel,
|
||||
panelType: queryArg.panelType,
|
||||
dataSourceType: queryArg.dataSourceType,
|
||||
permission: queryArg.permission,
|
||||
sort: queryArg.sort,
|
||||
limit: queryArg.limit,
|
||||
@@ -674,6 +676,10 @@ export type SearchDashboardsAndFoldersApiArg = {
|
||||
tags?: string[];
|
||||
/** find dashboards that reference a given libraryPanel */
|
||||
libraryPanel?: string;
|
||||
/** find dashboards using panels of a given plugin type */
|
||||
panelType?: string;
|
||||
/** find dashboards using datasources of a given plugin type */
|
||||
dataSourceType?: string;
|
||||
/** permission needed for the resource (view, edit, admin) */
|
||||
permission?: 'view' | 'edit' | 'admin';
|
||||
/** sortable field */
|
||||
|
||||
@@ -657,10 +657,6 @@ export interface FeatureToggles {
|
||||
*/
|
||||
rolePickerDrawer?: boolean;
|
||||
/**
|
||||
* Enable unified storage search
|
||||
*/
|
||||
unifiedStorageSearch?: boolean;
|
||||
/**
|
||||
* Enable sprinkles on unified storage search
|
||||
*/
|
||||
unifiedStorageSearchSprinkles?: boolean;
|
||||
|
||||
@@ -52,6 +52,7 @@ export const availableIconsIndex = {
|
||||
bookmark: true,
|
||||
'book-open': true,
|
||||
'brackets-curly': true,
|
||||
brain: true,
|
||||
'browser-alt': true,
|
||||
bug: true,
|
||||
building: true,
|
||||
|
||||
@@ -74,7 +74,7 @@ func ToUnifiedStorage(c utils.CommandLine, cfg *setting.Cfg, sqlStore db.DB) err
|
||||
return err
|
||||
}
|
||||
|
||||
grpcClient, err := newUnifiedClient(cfg, sqlStore, featureToggles)
|
||||
grpcClient, err := newUnifiedMigratorClient(cfg, sqlStore, featureToggles)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -92,7 +92,7 @@ func ToUnifiedStorage(c utils.CommandLine, cfg *setting.Cfg, sqlStore db.DB) err
|
||||
return runInteractiveMigration(ctx, cfg, opts, dashboardAccess, grpcClient, start)
|
||||
}
|
||||
|
||||
func runNonInteractiveMigration(ctx context.Context, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.ResourceClient, start time.Time) error {
|
||||
func runNonInteractiveMigration(ctx context.Context, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.MigratorClient, start time.Time) error {
|
||||
migrator := migrations.ProvideUnifiedMigrator(dashboardAccess, grpcClient)
|
||||
|
||||
opts.WithHistory = true // always include history in non-interactive mode
|
||||
@@ -109,7 +109,7 @@ func runNonInteractiveMigration(ctx context.Context, opts legacy.MigrateOptions,
|
||||
return nil
|
||||
}
|
||||
|
||||
func runInteractiveMigration(ctx context.Context, cfg *setting.Cfg, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.ResourceClient, start time.Time) error {
|
||||
func runInteractiveMigration(ctx context.Context, cfg *setting.Cfg, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.MigratorClient, start time.Time) error {
|
||||
yes, err := promptYesNo(fmt.Sprintf("Count legacy resources for namespace: %s?", opts.Namespace))
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -225,7 +225,7 @@ func promptYesNo(prompt string) (bool, error) {
|
||||
}
|
||||
}
|
||||
|
||||
func newUnifiedClient(cfg *setting.Cfg, sqlStore db.DB, featureToggles featuremgmt.FeatureToggles) (resource.ResourceClient, error) {
|
||||
func newUnifiedMigratorClient(cfg *setting.Cfg, sqlStore db.DB, featureToggles featuremgmt.FeatureToggles) (resource.MigratorClient, error) {
|
||||
return unified.ProvideUnifiedStorageClient(&unified.Options{
|
||||
Cfg: cfg,
|
||||
Features: featureToggles,
|
||||
|
||||
@@ -10,6 +10,7 @@ const (
|
||||
SearchServerRing string = "search-server-ring"
|
||||
SearchServerDistributor string = "search-server-distributor"
|
||||
StorageServer string = "storage-server"
|
||||
SearchServer string = "search-server"
|
||||
ZanzanaServer string = "zanzana-server"
|
||||
InstrumentationServer string = "instrumentation-server"
|
||||
FrontendServer string = "frontend-server"
|
||||
@@ -21,6 +22,7 @@ var dependencyMap = map[string][]string{
|
||||
SearchServerRing: {InstrumentationServer, MemberlistKV},
|
||||
GrafanaAPIServer: {InstrumentationServer},
|
||||
StorageServer: {InstrumentationServer, SearchServerRing},
|
||||
SearchServer: {InstrumentationServer, SearchServerRing},
|
||||
ZanzanaServer: {InstrumentationServer},
|
||||
SearchServerDistributor: {InstrumentationServer, MemberlistKV, SearchServerRing},
|
||||
Core: {},
|
||||
|
||||
@@ -11,91 +11,59 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
_ resource.ResourceClient = (*directResourceClient)(nil)
|
||||
_ resource.StorageClient = (*DirectStorageClient)(nil)
|
||||
)
|
||||
|
||||
// The direct client passes requests directly to the server using the *same* context
|
||||
func NewDirectResourceClient(server resource.ResourceServer) resource.ResourceClient {
|
||||
return &directResourceClient{server}
|
||||
// NewDirectStorageClient creates a client that passes requests directly to the server using the *same* context
|
||||
func NewDirectStorageClient(server resource.ResourceServer) *DirectStorageClient {
|
||||
return &DirectStorageClient{server}
|
||||
}
|
||||
|
||||
type directResourceClient struct {
|
||||
type DirectStorageClient struct {
|
||||
server resource.ResourceServer
|
||||
}
|
||||
|
||||
// Create implements ResourceClient.
|
||||
func (d *directResourceClient) Create(ctx context.Context, in *resourcepb.CreateRequest, opts ...grpc.CallOption) (*resourcepb.CreateResponse, error) {
|
||||
func (d *DirectStorageClient) Create(ctx context.Context, in *resourcepb.CreateRequest, _ ...grpc.CallOption) (*resourcepb.CreateResponse, error) {
|
||||
return d.server.Create(ctx, in)
|
||||
}
|
||||
|
||||
// Delete implements ResourceClient.
|
||||
func (d *directResourceClient) Delete(ctx context.Context, in *resourcepb.DeleteRequest, opts ...grpc.CallOption) (*resourcepb.DeleteResponse, error) {
|
||||
func (d *DirectStorageClient) Delete(ctx context.Context, in *resourcepb.DeleteRequest, _ ...grpc.CallOption) (*resourcepb.DeleteResponse, error) {
|
||||
return d.server.Delete(ctx, in)
|
||||
}
|
||||
|
||||
// GetBlob implements ResourceClient.
|
||||
func (d *directResourceClient) GetBlob(ctx context.Context, in *resourcepb.GetBlobRequest, opts ...grpc.CallOption) (*resourcepb.GetBlobResponse, error) {
|
||||
func (d *DirectStorageClient) GetBlob(ctx context.Context, in *resourcepb.GetBlobRequest, _ ...grpc.CallOption) (*resourcepb.GetBlobResponse, error) {
|
||||
return d.server.GetBlob(ctx, in)
|
||||
}
|
||||
|
||||
// GetStats implements ResourceClient.
|
||||
func (d *directResourceClient) GetStats(ctx context.Context, in *resourcepb.ResourceStatsRequest, opts ...grpc.CallOption) (*resourcepb.ResourceStatsResponse, error) {
|
||||
return d.server.GetStats(ctx, in)
|
||||
}
|
||||
|
||||
// IsHealthy implements ResourceClient.
|
||||
func (d *directResourceClient) IsHealthy(ctx context.Context, in *resourcepb.HealthCheckRequest, opts ...grpc.CallOption) (*resourcepb.HealthCheckResponse, error) {
|
||||
func (d *DirectStorageClient) IsHealthy(ctx context.Context, in *resourcepb.HealthCheckRequest, _ ...grpc.CallOption) (*resourcepb.HealthCheckResponse, error) {
|
||||
return d.server.IsHealthy(ctx, in)
|
||||
}
|
||||
|
||||
// List implements ResourceClient.
|
||||
func (d *directResourceClient) List(ctx context.Context, in *resourcepb.ListRequest, opts ...grpc.CallOption) (*resourcepb.ListResponse, error) {
|
||||
func (d *DirectStorageClient) List(ctx context.Context, in *resourcepb.ListRequest, _ ...grpc.CallOption) (*resourcepb.ListResponse, error) {
|
||||
return d.server.List(ctx, in)
|
||||
}
|
||||
|
||||
func (d *directResourceClient) ListManagedObjects(ctx context.Context, in *resourcepb.ListManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.ListManagedObjectsResponse, error) {
|
||||
return d.server.ListManagedObjects(ctx, in)
|
||||
}
|
||||
|
||||
func (d *directResourceClient) CountManagedObjects(ctx context.Context, in *resourcepb.CountManagedObjectsRequest, opts ...grpc.CallOption) (*resourcepb.CountManagedObjectsResponse, error) {
|
||||
return d.server.CountManagedObjects(ctx, in)
|
||||
}
|
||||
|
||||
// PutBlob implements ResourceClient.
|
||||
func (d *directResourceClient) PutBlob(ctx context.Context, in *resourcepb.PutBlobRequest, opts ...grpc.CallOption) (*resourcepb.PutBlobResponse, error) {
|
||||
func (d *DirectStorageClient) PutBlob(ctx context.Context, in *resourcepb.PutBlobRequest, _ ...grpc.CallOption) (*resourcepb.PutBlobResponse, error) {
|
||||
return d.server.PutBlob(ctx, in)
|
||||
}
|
||||
|
||||
// Read implements ResourceClient.
|
||||
func (d *directResourceClient) Read(ctx context.Context, in *resourcepb.ReadRequest, opts ...grpc.CallOption) (*resourcepb.ReadResponse, error) {
|
||||
func (d *DirectStorageClient) Read(ctx context.Context, in *resourcepb.ReadRequest, _ ...grpc.CallOption) (*resourcepb.ReadResponse, error) {
|
||||
return d.server.Read(ctx, in)
|
||||
}
|
||||
|
||||
// Search implements ResourceClient.
|
||||
func (d *directResourceClient) Search(ctx context.Context, in *resourcepb.ResourceSearchRequest, opts ...grpc.CallOption) (*resourcepb.ResourceSearchResponse, error) {
|
||||
return d.server.Search(ctx, in)
|
||||
}
|
||||
|
||||
// Update implements ResourceClient.
|
||||
func (d *directResourceClient) Update(ctx context.Context, in *resourcepb.UpdateRequest, opts ...grpc.CallOption) (*resourcepb.UpdateResponse, error) {
|
||||
func (d *DirectStorageClient) Update(ctx context.Context, in *resourcepb.UpdateRequest, _ ...grpc.CallOption) (*resourcepb.UpdateResponse, error) {
|
||||
return d.server.Update(ctx, in)
|
||||
}
|
||||
|
||||
// Watch implements ResourceClient.
|
||||
func (d *directResourceClient) Watch(ctx context.Context, in *resourcepb.WatchRequest, opts ...grpc.CallOption) (resourcepb.ResourceStore_WatchClient, error) {
|
||||
func (d *DirectStorageClient) Watch(_ context.Context, _ *resourcepb.WatchRequest, _ ...grpc.CallOption) (resourcepb.ResourceStore_WatchClient, error) {
|
||||
return nil, fmt.Errorf("watch not supported with direct resource client")
|
||||
}
|
||||
|
||||
// BulkProcess implements resource.ResourceClient.
|
||||
func (d *directResourceClient) BulkProcess(ctx context.Context, opts ...grpc.CallOption) (resourcepb.BulkStore_BulkProcessClient, error) {
|
||||
return nil, fmt.Errorf("BulkProcess not supported with direct resource client")
|
||||
}
|
||||
|
||||
// RebuildIndexes implements resource.ResourceClient.
|
||||
func (b *directResourceClient) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest, opts ...grpc.CallOption) (*resourcepb.RebuildIndexesResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
func (b *directResourceClient) GetQuotaUsage(ctx context.Context, req *resourcepb.QuotaUsageRequest, opts ...grpc.CallOption) (*resourcepb.QuotaUsageResponse, error) {
|
||||
return nil, fmt.Errorf("not implemented")
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
|
||||
"github.com/grafana/authlib/types"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
grafanaregistry "github.com/grafana/grafana/pkg/apiserver/registry/generic"
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
@@ -40,7 +41,7 @@ func (s *DashboardStorage) NewStore(dash utils.ResourceInfo, scheme *runtime.Sch
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
client := legacy.NewDirectResourceClient(server) // same context
|
||||
client := legacy.NewDirectStorageClient(server) // same context
|
||||
optsGetter := apistore.NewRESTOptionsGetterForClient(client, nil,
|
||||
defaultOpts.StorageConfig.Config, nil,
|
||||
)
|
||||
|
||||
@@ -142,6 +142,24 @@ func (s *SearchHandler) GetAPIRoutes(defs map[string]common.OpenAPIDefinition) *
|
||||
Schema: spec.StringProperty(),
|
||||
},
|
||||
},
|
||||
{
|
||||
ParameterProps: spec3.ParameterProps{
|
||||
Name: "panelType",
|
||||
In: "query",
|
||||
Description: "find dashboards using panels of a given plugin type",
|
||||
Required: false,
|
||||
Schema: spec.StringProperty(),
|
||||
},
|
||||
},
|
||||
{
|
||||
ParameterProps: spec3.ParameterProps{
|
||||
Name: "dataSourceType",
|
||||
In: "query",
|
||||
Description: "find dashboards using datasources of a given plugin type",
|
||||
Required: false,
|
||||
Schema: spec.StringProperty(),
|
||||
},
|
||||
},
|
||||
{
|
||||
ParameterProps: spec3.ParameterProps{
|
||||
Name: "permission",
|
||||
@@ -430,14 +448,11 @@ func convertHttpSearchRequestToResourceSearchRequest(queryParams url.Values, use
|
||||
}
|
||||
}
|
||||
|
||||
// The facet term fields
|
||||
// Apply facet terms
|
||||
if facets, ok := queryParams["facet"]; ok {
|
||||
if queryParams.Has("facetLimit") {
|
||||
if parsed, err := strconv.Atoi(queryParams.Get("facetLimit")); err == nil && parsed > 0 {
|
||||
facetLimit = parsed
|
||||
if facetLimit > 1000 {
|
||||
facetLimit = 1000
|
||||
}
|
||||
facetLimit = min(parsed, 1000)
|
||||
}
|
||||
}
|
||||
searchRequest.Facet = make(map[string]*resourcepb.ResourceSearchRequest_Facet)
|
||||
@@ -449,21 +464,35 @@ func convertHttpSearchRequestToResourceSearchRequest(queryParams url.Values, use
|
||||
}
|
||||
}
|
||||
|
||||
// The tags filter
|
||||
if tags, ok := queryParams["tag"]; ok {
|
||||
if v, ok := queryParams["tag"]; ok {
|
||||
searchRequest.Options.Fields = append(searchRequest.Options.Fields, &resourcepb.Requirement{
|
||||
Key: "tags",
|
||||
Operator: "=",
|
||||
Values: tags,
|
||||
Values: v,
|
||||
})
|
||||
}
|
||||
|
||||
// The libraryPanel filter
|
||||
if libraryPanel, ok := queryParams["libraryPanel"]; ok {
|
||||
if v, ok := queryParams["panelType"]; ok {
|
||||
searchRequest.Options.Fields = append(searchRequest.Options.Fields, &resourcepb.Requirement{
|
||||
Key: resource.SEARCH_FIELD_PREFIX + builders.DASHBOARD_PANEL_TYPES,
|
||||
Operator: "=",
|
||||
Values: v,
|
||||
})
|
||||
}
|
||||
|
||||
if v, ok := queryParams["dataSourceType"]; ok {
|
||||
searchRequest.Options.Fields = append(searchRequest.Options.Fields, &resourcepb.Requirement{
|
||||
Key: resource.SEARCH_FIELD_PREFIX + builders.DASHBOARD_DS_TYPES,
|
||||
Operator: "=",
|
||||
Values: v,
|
||||
})
|
||||
}
|
||||
|
||||
if v, ok := queryParams["libraryPanel"]; ok {
|
||||
searchRequest.Options.Fields = append(searchRequest.Options.Fields, &resourcepb.Requirement{
|
||||
Key: builders.DASHBOARD_LIBRARY_PANEL_REFERENCE,
|
||||
Operator: "=",
|
||||
Values: libraryPanel,
|
||||
Values: v,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -37,7 +37,6 @@ import (
|
||||
"github.com/grafana/grafana/pkg/services/folder"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/apistore"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||
)
|
||||
|
||||
@@ -74,7 +73,7 @@ func RegisterAPIService(cfg *setting.Cfg,
|
||||
acService accesscontrol.Service,
|
||||
accessClient authlib.AccessClient,
|
||||
registerer prometheus.Registerer,
|
||||
unified resource.ResourceClient,
|
||||
unified resourcepb.ResourceIndexClient,
|
||||
zanzanaClient zanzana.Client,
|
||||
) *FolderAPIBuilder {
|
||||
builder := &FolderAPIBuilder{
|
||||
@@ -93,7 +92,7 @@ func RegisterAPIService(cfg *setting.Cfg,
|
||||
return builder
|
||||
}
|
||||
|
||||
func NewAPIService(ac authlib.AccessClient, searcher resource.ResourceClient, features featuremgmt.FeatureToggles, zanzanaClient zanzana.Client, resourcePermissionsSvc *dynamic.NamespaceableResourceInterface) *FolderAPIBuilder {
|
||||
func NewAPIService(ac authlib.AccessClient, searcher resourcepb.ResourceIndexClient, features featuremgmt.FeatureToggles, zanzanaClient zanzana.Client, resourcePermissionsSvc *dynamic.NamespaceableResourceInterface) *FolderAPIBuilder {
|
||||
return &FolderAPIBuilder{
|
||||
features: features,
|
||||
accessClient: ac,
|
||||
|
||||
@@ -742,7 +742,7 @@ func NewLocalStore(resourceInfo utils.ResourceInfo, scheme *runtime.Scheme, defa
|
||||
return nil, err
|
||||
}
|
||||
|
||||
client := resource.NewLocalResourceClient(server)
|
||||
client := resource.NewLocalResourceClient(server, nil)
|
||||
optsGetter := apistore.NewRESTOptionsGetterForClient(client, nil, defaultOpts.StorageConfig.Config, nil)
|
||||
|
||||
store, err := grafanaregistry.NewRegistryStore(scheme, resourceInfo, optsGetter)
|
||||
|
||||
@@ -205,6 +205,14 @@ func (s *ModuleServer) Run() error {
|
||||
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)
|
||||
})
|
||||
|
||||
m.RegisterModule(modules.SearchServer, func() (services.Service, error) {
|
||||
docBuilders, err := InitializeDocumentBuilders(s.cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return sql.ProvideUnifiedSearchGrpcService(s.cfg, s.features, nil, s.log, s.registerer, docBuilders, s.indexMetrics, s.searchServerRing, s.MemberlistKVConfig, s.storageBackend)
|
||||
})
|
||||
|
||||
m.RegisterModule(modules.ZanzanaServer, func() (services.Service, error) {
|
||||
return authz.ProvideZanzanaService(s.cfg, s.features, s.registerer)
|
||||
})
|
||||
|
||||
@@ -372,7 +372,7 @@ func initModuleServerForTest(
|
||||
return testModuleServer{server: ms, grpcAddress: cfg.GRPCServer.Address, httpPort: cfg.HTTPPort, healthClient: healthClient, id: cfg.InstanceID}
|
||||
}
|
||||
|
||||
func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces []string) resource.ResourceServer {
|
||||
func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces []string) resource.SearchServer {
|
||||
cfg := setting.NewCfg()
|
||||
section, err := cfg.Raw.NewSection("database")
|
||||
require.NoError(t, err)
|
||||
@@ -391,17 +391,20 @@ func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces
|
||||
require.NoError(t, err)
|
||||
searchOpts, err := search.NewSearchOptions(features, cfg, docBuilders, nil, nil)
|
||||
require.NoError(t, err)
|
||||
server, err := sql.NewResourceServer(sql.ServerOptions{
|
||||
DB: nil,
|
||||
Cfg: cfg,
|
||||
Tracer: tracer,
|
||||
Reg: nil,
|
||||
AccessClient: nil,
|
||||
SearchOptions: searchOpts,
|
||||
StorageMetrics: nil,
|
||||
IndexMetrics: nil,
|
||||
Features: features,
|
||||
QOSQueue: nil,
|
||||
searchServer, err := sql.NewSearchServer(sql.SearchServerOptions{
|
||||
DB: nil,
|
||||
Cfg: cfg,
|
||||
Tracer: tracer,
|
||||
Reg: nil,
|
||||
AccessClient: nil,
|
||||
SearchOptions: searchOpts,
|
||||
IndexMetrics: nil,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
storageServer, err := sql.NewStorageServer(sql.StorageServerOptions{
|
||||
Cfg: cfg,
|
||||
Tracer: tracer,
|
||||
Features: features,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -417,12 +420,12 @@ func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces
|
||||
|
||||
for _, ns := range testNamespaces {
|
||||
for range rand.Intn(maxPlaylistPerNamespace) + 1 {
|
||||
_, err = server.Create(ctx, generatePlaylistPayload(ns))
|
||||
_, err = storageServer.Create(ctx, generatePlaylistPayload(ns))
|
||||
require.NoError(t, err)
|
||||
}
|
||||
}
|
||||
|
||||
return server
|
||||
return searchServer
|
||||
}
|
||||
|
||||
var counter int
|
||||
|
||||
Generated
+20
-3
@@ -513,6 +513,11 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
storageMetrics := resource.ProvideStorageMetrics(registerer)
|
||||
storageBackend, err := sql.ProvideStorageBackend(cfg, sqlStore, tracer, registerer, storageMetrics)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
options := &unified.Options{
|
||||
Cfg: cfg,
|
||||
Features: featureToggles,
|
||||
@@ -522,8 +527,8 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
|
||||
Authzc: accessClient,
|
||||
Docs: documentBuilderSupplier,
|
||||
SecureValues: inlineSecureValueSupport,
|
||||
Backend: storageBackend,
|
||||
}
|
||||
storageMetrics := resource.ProvideStorageMetrics(registerer)
|
||||
bleveIndexMetrics := resource.ProvideIndexMetrics(registerer)
|
||||
resourceClient, err := unified.ProvideUnifiedStorageClient(options, storageMetrics, bleveIndexMetrics)
|
||||
if err != nil {
|
||||
@@ -1173,6 +1178,11 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
storageMetrics := resource.ProvideStorageMetrics(registerer)
|
||||
storageBackend, err := sql.ProvideStorageBackend(cfg, sqlStore, tracer, registerer, storageMetrics)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
options := &unified.Options{
|
||||
Cfg: cfg,
|
||||
Features: featureToggles,
|
||||
@@ -1182,8 +1192,8 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
|
||||
Authzc: accessClient,
|
||||
Docs: documentBuilderSupplier,
|
||||
SecureValues: inlineSecureValueSupport,
|
||||
Backend: storageBackend,
|
||||
}
|
||||
storageMetrics := resource.ProvideStorageMetrics(registerer)
|
||||
bleveIndexMetrics := resource.ProvideIndexMetrics(registerer)
|
||||
resourceClient, err := unified.ProvideUnifiedStorageClient(options, storageMetrics, bleveIndexMetrics)
|
||||
if err != nil {
|
||||
@@ -1748,7 +1758,14 @@ func InitializeModuleServer(cfg *setting.Cfg, opts Options, apiOpts api.ServerOp
|
||||
hooksService := hooks.ProvideService()
|
||||
ossLicensingService := licensing.ProvideService(cfg, hooksService)
|
||||
moduleRegisterer := ProvideNoopModuleRegisterer()
|
||||
storageBackend, err := sql.ProvideStorageBackend(cfg)
|
||||
ossMigrations := migrations.ProvideOSSMigrations(featureToggles)
|
||||
inProcBus := bus.ProvideBus(tracingService)
|
||||
sqlStore, err := sqlstore.ProvideService(cfg, featureToggles, ossMigrations, inProcBus, tracingService)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
tracer := otelTracer()
|
||||
storageBackend, err := sql.ProvideStorageBackend(cfg, sqlStore, tracer, registerer, storageMetrics)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -6,6 +6,9 @@ package server
|
||||
|
||||
import (
|
||||
"github.com/google/wire"
|
||||
"github.com/grafana/grafana/pkg/bus"
|
||||
"github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/services/sqlstore"
|
||||
|
||||
"github.com/grafana/grafana/pkg/configprovider"
|
||||
"github.com/grafana/grafana/pkg/infra/metrics"
|
||||
@@ -65,6 +68,7 @@ import (
|
||||
"github.com/grafana/grafana/pkg/storage/legacysql"
|
||||
"github.com/grafana/grafana/pkg/storage/unified"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||
search2 "github.com/grafana/grafana/pkg/storage/unified/search"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/search/builders"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/sql"
|
||||
@@ -145,8 +149,10 @@ var wireExtsBasicSet = wire.NewSet(
|
||||
sandbox.ProvideService,
|
||||
wire.Bind(new(sandbox.Sandbox), new(*sandbox.Service)),
|
||||
wire.Struct(new(unified.Options), "*"),
|
||||
unified.ProvideUnifiedStorageClient,
|
||||
sql.ProvideStorageBackend,
|
||||
unified.ProvideUnifiedStorageClient,
|
||||
wire.Bind(new(resourcepb.ResourceIndexClient), new(resource.ResourceClient)),
|
||||
wire.Bind(new(resource.MigratorClient), new(resource.ResourceClient)),
|
||||
builder.ProvideDefaultBuildHandlerChainFuncFromBuilders,
|
||||
aggregatorrunner.ProvideNoopAggregatorConfigurator,
|
||||
apisregistry.WireSetExts,
|
||||
@@ -195,6 +201,16 @@ var wireExtsModuleServerSet = wire.NewSet(
|
||||
tracing.ProvideTracingConfig,
|
||||
tracing.ProvideService,
|
||||
wire.Bind(new(tracing.Tracer), new(*tracing.TracingService)),
|
||||
otelTracer,
|
||||
// Bus
|
||||
bus.ProvideBus,
|
||||
wire.Bind(new(bus.Bus), new(*bus.InProcBus)),
|
||||
// Database migrations
|
||||
migrations.ProvideOSSMigrations,
|
||||
wire.Bind(new(registry.DatabaseMigrator), new(*migrations.OSSMigrations)),
|
||||
// Database
|
||||
sqlstore.ProvideService,
|
||||
wire.Bind(new(db.DB), new(*sqlstore.SQLStore)),
|
||||
// Unified storage
|
||||
resource.ProvideStorageMetrics,
|
||||
resource.ProvideIndexMetrics,
|
||||
|
||||
@@ -1087,13 +1087,6 @@ var (
|
||||
Stage: FeatureStageExperimental,
|
||||
Owner: identityAccessTeam,
|
||||
},
|
||||
{
|
||||
Name: "unifiedStorageSearch",
|
||||
Description: "Enable unified storage search",
|
||||
Stage: FeatureStageExperimental,
|
||||
Owner: grafanaSearchAndStorageSquad,
|
||||
HideFromDocs: true,
|
||||
},
|
||||
{
|
||||
Name: "unifiedStorageSearchSprinkles",
|
||||
Description: "Enable sprinkles on unified storage search",
|
||||
|
||||
Generated
-1
@@ -150,7 +150,6 @@ alertingQueryAndExpressionsStepMode,GA,@grafana/alerting-squad,false,false,true
|
||||
improvedExternalSessionHandling,GA,@grafana/identity-access-team,false,false,false
|
||||
useSessionStorageForRedirection,GA,@grafana/identity-access-team,false,false,false
|
||||
rolePickerDrawer,experimental,@grafana/identity-access-team,false,false,false
|
||||
unifiedStorageSearch,experimental,@grafana/search-and-storage,false,false,false
|
||||
unifiedStorageSearchSprinkles,experimental,@grafana/search-and-storage,false,false,false
|
||||
managedDualWriter,experimental,@grafana/search-and-storage,false,false,false
|
||||
pluginsSriChecks,GA,@grafana/plugins-platform-backend,false,false,false
|
||||
|
||||
|
Generated
-4
@@ -455,10 +455,6 @@ const (
|
||||
// Enables the new role picker drawer design
|
||||
FlagRolePickerDrawer = "rolePickerDrawer"
|
||||
|
||||
// FlagUnifiedStorageSearch
|
||||
// Enable unified storage search
|
||||
FlagUnifiedStorageSearch = "unifiedStorageSearch"
|
||||
|
||||
// FlagUnifiedStorageSearchSprinkles
|
||||
// Enable sprinkles on unified storage search
|
||||
FlagUnifiedStorageSearchSprinkles = "unifiedStorageSearchSprinkles"
|
||||
|
||||
+2
-1
@@ -3697,7 +3697,8 @@
|
||||
"metadata": {
|
||||
"name": "unifiedStorageSearch",
|
||||
"resourceVersion": "1764664939750",
|
||||
"creationTimestamp": "2024-09-30T19:46:14Z"
|
||||
"creationTimestamp": "2024-09-30T19:46:14Z",
|
||||
"deletionTimestamp": "2026-01-12T10:02:12Z"
|
||||
},
|
||||
"spec": {
|
||||
"description": "Enable unified storage search",
|
||||
|
||||
@@ -100,6 +100,9 @@ func (d *DsLookup) ByRef(ref *DataSourceRef) *DataSourceRef {
|
||||
if ref == nil {
|
||||
return d.defaultDS
|
||||
}
|
||||
if ref.UID == "default" && ref.Type == "" {
|
||||
return d.defaultDS
|
||||
}
|
||||
|
||||
key := ""
|
||||
if ref.UID != "" {
|
||||
@@ -117,7 +120,13 @@ func (d *DsLookup) ByRef(ref *DataSourceRef) *DataSourceRef {
|
||||
return ds
|
||||
}
|
||||
|
||||
return d.byName[key]
|
||||
ds, ok = d.byName[key]
|
||||
if ok {
|
||||
return ds
|
||||
}
|
||||
|
||||
// With nothing was found (or configured), use the original reference
|
||||
return ref
|
||||
}
|
||||
|
||||
func (d *DsLookup) ByType(dsType string) []DataSourceRef {
|
||||
|
||||
+4
-4
@@ -4,8 +4,8 @@
|
||||
"tags": null,
|
||||
"datasource": [
|
||||
{
|
||||
"uid": "default.uid",
|
||||
"type": "default.type"
|
||||
"uid": "000000001",
|
||||
"type": "graphite"
|
||||
}
|
||||
],
|
||||
"panels": [
|
||||
@@ -16,8 +16,8 @@
|
||||
"libraryPanel": "dfkljg98345dkf",
|
||||
"datasource": [
|
||||
{
|
||||
"uid": "default.uid",
|
||||
"type": "default.type"
|
||||
"uid": "000000001",
|
||||
"type": "graphite"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
package dashboard
|
||||
|
||||
import "iter"
|
||||
|
||||
type PanelSummaryInfo struct {
|
||||
ID int64 `json:"id"`
|
||||
Title string `json:"title"`
|
||||
@@ -30,3 +32,20 @@ type DashboardSummaryInfo struct {
|
||||
Refresh string `json:"refresh,omitempty"`
|
||||
ReadOnly bool `json:"readOnly,omitempty"` // editable = false
|
||||
}
|
||||
|
||||
func (d *DashboardSummaryInfo) PanelIterator() iter.Seq[PanelSummaryInfo] {
|
||||
return func(yield func(PanelSummaryInfo) bool) {
|
||||
for _, p := range d.Panels {
|
||||
if len(p.Collapsed) > 0 {
|
||||
for _, c := range p.Collapsed {
|
||||
if !yield(c) { // NOTE, rows can only be one level deep!
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
if !yield(p) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -236,7 +236,6 @@ kubernetesDashboards = true
|
||||
kubernetesFolders = true
|
||||
unifiedStorage = true
|
||||
unifiedStorageHistoryPruner = true
|
||||
unifiedStorageSearch = true
|
||||
unifiedStorageSearchPermissionFiltering = false
|
||||
unifiedStorageSearchSprinkles = false
|
||||
|
||||
|
||||
@@ -26,7 +26,7 @@ var _ generic.RESTOptionsGetter = (*RESTOptionsGetter)(nil)
|
||||
type StorageOptionsRegister func(gr schema.GroupResource, opts StorageOptions)
|
||||
|
||||
type RESTOptionsGetter struct {
|
||||
client resource.ResourceClient
|
||||
client resource.StorageClient
|
||||
secrets secret.InlineSecureValueSupport
|
||||
original storagebackend.Config
|
||||
configProvider RestConfigProvider
|
||||
@@ -36,7 +36,7 @@ type RESTOptionsGetter struct {
|
||||
}
|
||||
|
||||
func NewRESTOptionsGetterForClient(
|
||||
client resource.ResourceClient,
|
||||
client resource.StorageClient,
|
||||
secrets secret.InlineSecureValueSupport,
|
||||
original storagebackend.Config,
|
||||
configProvider RestConfigProvider,
|
||||
@@ -79,7 +79,7 @@ func NewRESTOptionsGetterMemory(originalStorageConfig storagebackend.Config, sec
|
||||
}
|
||||
|
||||
return NewRESTOptionsGetterForClient(
|
||||
resource.NewLocalResourceClient(server),
|
||||
resource.NewLocalResourceClient(server, nil),
|
||||
secrets,
|
||||
originalStorageConfig,
|
||||
nil,
|
||||
@@ -118,7 +118,7 @@ func NewRESTOptionsGetterForFileXX(path string,
|
||||
}
|
||||
|
||||
return NewRESTOptionsGetterForClient(
|
||||
resource.NewLocalResourceClient(server),
|
||||
resource.NewLocalResourceClient(server, nil),
|
||||
nil, // secrets
|
||||
originalStorageConfig,
|
||||
nil,
|
||||
|
||||
@@ -88,7 +88,7 @@ type Storage struct {
|
||||
trigger storage.IndexerFuncs
|
||||
indexers *cache.Indexers
|
||||
|
||||
store resource.ResourceClient
|
||||
store resource.StorageClient
|
||||
getKey func(string) (*resourcepb.ResourceKey, error)
|
||||
snowflake *snowflake.Node // used to enforce internal ids
|
||||
configProvider RestConfigProvider // used for provisioning
|
||||
@@ -112,7 +112,7 @@ type RestConfigProvider interface {
|
||||
// NewStorage instantiates a new Storage.
|
||||
func NewStorage(
|
||||
config *storagebackend.ConfigForResource,
|
||||
store resource.ResourceClient,
|
||||
store resource.StorageClient,
|
||||
keyFunc func(obj runtime.Object) (string, error),
|
||||
keyParser func(key string) (*resourcepb.ResourceKey, error),
|
||||
newFunc func() runtime.Object,
|
||||
|
||||
@@ -156,7 +156,7 @@ func testSetup(t testing.TB, opts ...setupOption) (context.Context, storage.Inte
|
||||
default:
|
||||
t.Fatalf("unsupported storage type: %s", setupOpts.storageType)
|
||||
}
|
||||
client := resource.NewLocalResourceClient(server)
|
||||
client := resource.NewLocalResourceClient(server, nil)
|
||||
|
||||
config := storagebackend.NewDefaultConfig(setupOpts.prefix, setupOpts.codec)
|
||||
store, destroyFunc, err := apistore.NewStorage(
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"github.com/grafana/dskit/grpcclient"
|
||||
"github.com/grafana/dskit/middleware"
|
||||
"github.com/grafana/dskit/services"
|
||||
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
infraDB "github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
@@ -45,6 +46,7 @@ type Options struct {
|
||||
Authzc types.AccessClient
|
||||
Docs resource.DocumentBuilderSupplier
|
||||
SecureValues secrets.InlineSecureValueSupport
|
||||
Backend resource.StorageBackend // Shared backend to avoid duplicate metrics registration
|
||||
}
|
||||
|
||||
type clientMetrics struct {
|
||||
@@ -66,7 +68,7 @@ func ProvideUnifiedStorageClient(opts *Options,
|
||||
BlobStoreURL: apiserverCfg.Key("blob_url").MustString(""),
|
||||
BlobThresholdBytes: apiserverCfg.Key("blob_threshold_bytes").MustInt(options.BlobThresholdDefault),
|
||||
GrpcClientKeepaliveTime: apiserverCfg.Key("grpc_client_keepalive_time").MustDuration(0),
|
||||
}, opts.Cfg, opts.Features, opts.DB, opts.Tracer, opts.Reg, opts.Authzc, opts.Docs, storageMetrics, indexMetrics, opts.SecureValues)
|
||||
}, opts.Cfg, opts.Features, opts.DB, opts.Tracer, opts.Reg, opts.Authzc, opts.Docs, storageMetrics, indexMetrics, opts.SecureValues, opts.Backend)
|
||||
if err == nil {
|
||||
// Decide whether to disable SQL fallback stats per resource in Mode 5.
|
||||
// Otherwise we would still try to query the legacy SQL database in Mode 5.
|
||||
@@ -102,6 +104,7 @@ func newClient(opts options.StorageOptions,
|
||||
storageMetrics *resource.StorageMetrics,
|
||||
indexMetrics *resource.BleveIndexMetrics,
|
||||
secure secrets.InlineSecureValueSupport,
|
||||
backend resource.StorageBackend,
|
||||
) (resource.ResourceClient, error) {
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -135,7 +138,7 @@ func newClient(opts options.StorageOptions,
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resource.NewLocalResourceClient(server), nil
|
||||
return resource.NewLocalResourceClient(server, nil), nil
|
||||
|
||||
case options.StorageTypeUnifiedGrpc:
|
||||
if opts.Address == "" {
|
||||
@@ -168,24 +171,14 @@ func newClient(opts options.StorageOptions,
|
||||
return resource.NewResourceClient(conn, indexConn, cfg, features, tracer)
|
||||
|
||||
default:
|
||||
// Create search options for the search server
|
||||
searchOptions, err := search.NewSearchOptions(features, cfg, docs, indexMetrics, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
serverOptions := sql.ServerOptions{
|
||||
DB: db,
|
||||
Cfg: cfg,
|
||||
Tracer: tracer,
|
||||
Reg: reg,
|
||||
AccessClient: authzc,
|
||||
SearchOptions: searchOptions,
|
||||
StorageMetrics: storageMetrics,
|
||||
IndexMetrics: indexMetrics,
|
||||
Features: features,
|
||||
SecureValues: secure,
|
||||
}
|
||||
|
||||
// Setup QOS queue if enabled
|
||||
var qosQueue sql.QOSEnqueueDequeuer
|
||||
if cfg.QOSEnabled {
|
||||
qosReg := prometheus.WrapRegistererWithPrefix("resource_server_qos_", reg)
|
||||
queue := scheduler.NewQueue(&scheduler.QueueOptions{
|
||||
@@ -196,7 +189,7 @@ func newClient(opts options.StorageOptions,
|
||||
if err := services.StartAndAwaitRunning(ctx, queue); err != nil {
|
||||
return nil, fmt.Errorf("failed to start queue: %w", err)
|
||||
}
|
||||
scheduler, err := scheduler.NewScheduler(queue, &scheduler.Config{
|
||||
sched, err := scheduler.NewScheduler(queue, &scheduler.Config{
|
||||
NumWorkers: cfg.QOSNumberWorker,
|
||||
Logger: cfg.Logger,
|
||||
})
|
||||
@@ -204,31 +197,59 @@ func newClient(opts options.StorageOptions,
|
||||
return nil, fmt.Errorf("failed to create scheduler: %w", err)
|
||||
}
|
||||
|
||||
err = services.StartAndAwaitRunning(ctx, scheduler)
|
||||
err = services.StartAndAwaitRunning(ctx, sched)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to start scheduler: %w", err)
|
||||
}
|
||||
serverOptions.QOSQueue = queue
|
||||
qosQueue = queue
|
||||
}
|
||||
|
||||
// only enable if an overrides file path is provided
|
||||
// Setup overrides service if enabled
|
||||
var overridesSvc *resource.OverridesService
|
||||
if cfg.OverridesFilePath != "" {
|
||||
overridesSvc, err := resource.NewOverridesService(ctx, cfg.Logger, reg, tracer, resource.ReloadOptions{
|
||||
overridesSvc, err = resource.NewOverridesService(ctx, cfg.Logger, reg, tracer, resource.ReloadOptions{
|
||||
FilePath: cfg.OverridesFilePath,
|
||||
ReloadPeriod: cfg.OverridesReloadInterval,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
serverOptions.OverridesService = overridesSvc
|
||||
}
|
||||
|
||||
server, err := sql.NewResourceServer(serverOptions)
|
||||
// Create the search server with shared backend
|
||||
searchServer, err := sql.NewSearchServer(sql.SearchServerOptions{
|
||||
Backend: backend, // Use shared backend to avoid duplicate metrics registration
|
||||
DB: db,
|
||||
Cfg: cfg,
|
||||
Tracer: tracer,
|
||||
Reg: reg,
|
||||
AccessClient: authzc,
|
||||
SearchOptions: searchOptions,
|
||||
IndexMetrics: indexMetrics,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resource.NewLocalResourceClient(server), nil
|
||||
|
||||
// Create the storage server with shared backend
|
||||
storageServer, err := sql.NewStorageServer(sql.StorageServerOptions{
|
||||
Backend: backend, // Use shared backend to avoid duplicate metrics registration
|
||||
DB: db,
|
||||
Cfg: cfg,
|
||||
Tracer: tracer,
|
||||
Reg: reg,
|
||||
AccessClient: authzc,
|
||||
StorageMetrics: storageMetrics,
|
||||
Features: features,
|
||||
QOSQueue: qosQueue,
|
||||
SecureValues: secure,
|
||||
OverridesService: overridesSvc,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return resource.NewLocalResourceClient(storageServer, searchServer), nil
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -48,7 +48,7 @@ func buildCollectionSettings(opts legacy.MigrateOptions) resource.BulkSettings {
|
||||
}
|
||||
|
||||
type resourceClientStreamProvider struct {
|
||||
client resource.ResourceClient
|
||||
client resource.MigratorClient
|
||||
}
|
||||
|
||||
func (r *resourceClientStreamProvider) createStream(ctx context.Context, opts legacy.MigrateOptions) (resourcepb.BulkStore_BulkProcessClient, error) {
|
||||
@@ -71,7 +71,7 @@ func (b *bulkStoreClientStreamProvider) createStream(ctx context.Context, opts l
|
||||
// This can migrate Folders, Dashboards and LibraryPanels
|
||||
func ProvideUnifiedMigrator(
|
||||
dashboardAccess legacy.MigrationDashboardAccessor,
|
||||
client resource.ResourceClient,
|
||||
client resource.MigratorClient,
|
||||
) UnifiedMigrator {
|
||||
return newUnifiedMigrator(
|
||||
dashboardAccess,
|
||||
|
||||
@@ -31,13 +31,31 @@ import (
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||
)
|
||||
|
||||
//go:generate mockery --name ResourceClient --structname MockResourceClient --inpackage --filename client_mock.go --with-expecter
|
||||
type ResourceClient interface {
|
||||
resourcepb.ResourceStoreClient
|
||||
// SearchClient is used to interact with unified search
|
||||
type SearchClient interface {
|
||||
resourcepb.ResourceIndexClient
|
||||
resourcepb.ManagedObjectIndexClient
|
||||
resourcepb.BulkStoreClient
|
||||
}
|
||||
|
||||
// StorageClient is used to interact with unified storage
|
||||
type StorageClient interface {
|
||||
resourcepb.ResourceStoreClient
|
||||
resourcepb.BlobStoreClient
|
||||
}
|
||||
|
||||
// MigratorClient is used to perform migrations to unified storage
|
||||
type MigratorClient interface {
|
||||
resourcepb.BulkStoreClient
|
||||
GetStats(ctx context.Context, in *resourcepb.ResourceStatsRequest, opts ...grpc.CallOption) (*resourcepb.ResourceStatsResponse, error)
|
||||
}
|
||||
|
||||
// ResourceClient combines all resource-related clients and should be avoided in favor of more specific interfaces when possible
|
||||
//
|
||||
//go:generate mockery --name ResourceClient --structname MockResourceClient --inpackage --filename client_mock.go --with-expecter
|
||||
type ResourceClient interface {
|
||||
StorageClient
|
||||
SearchClient
|
||||
MigratorClient
|
||||
resourcepb.DiagnosticsClient
|
||||
resourcepb.QuotasClient
|
||||
}
|
||||
@@ -92,16 +110,15 @@ func NewLegacyResourceClient(channel grpc.ClientConnInterface, indexChannel grpc
|
||||
return newResourceClient(cc, cci)
|
||||
}
|
||||
|
||||
func NewLocalResourceClient(server ResourceServer) ResourceClient {
|
||||
func NewLocalResourceClient(server ResourceServer, searchServer SearchServer) ResourceClient {
|
||||
// scenario: local in-proc
|
||||
channel := &inprocgrpc.Channel{}
|
||||
indexChannel := &inprocgrpc.Channel{}
|
||||
tracer := otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/resource")
|
||||
|
||||
grpcAuthInt := grpcutils.NewUnsafeAuthenticator(tracer)
|
||||
for _, desc := range []*grpc.ServiceDesc{
|
||||
&resourcepb.ResourceStore_ServiceDesc,
|
||||
&resourcepb.ResourceIndex_ServiceDesc,
|
||||
&resourcepb.ManagedObjectIndex_ServiceDesc,
|
||||
&resourcepb.BlobStore_ServiceDesc,
|
||||
&resourcepb.BulkStore_ServiceDesc,
|
||||
&resourcepb.Diagnostics_ServiceDesc,
|
||||
@@ -117,13 +134,31 @@ func NewLocalResourceClient(server ResourceServer) ResourceClient {
|
||||
)
|
||||
}
|
||||
|
||||
// Register search services on the index channel if searchServer is provided
|
||||
if searchServer != nil {
|
||||
for _, desc := range []*grpc.ServiceDesc{
|
||||
&resourcepb.ResourceIndex_ServiceDesc,
|
||||
&resourcepb.ManagedObjectIndex_ServiceDesc,
|
||||
} {
|
||||
indexChannel.RegisterService(
|
||||
grpchan.InterceptServer(
|
||||
desc,
|
||||
grpcAuth.UnaryServerInterceptor(grpcAuthInt),
|
||||
grpcAuth.StreamServerInterceptor(grpcAuthInt),
|
||||
),
|
||||
searchServer,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
clientInt := authnlib.NewGrpcClientInterceptor(
|
||||
ProvideInProcExchanger(),
|
||||
authnlib.WithClientInterceptorIDTokenExtractor(idTokenExtractor),
|
||||
)
|
||||
|
||||
cc := grpchan.InterceptClientConn(channel, clientInt.UnaryClientInterceptor, clientInt.StreamClientInterceptor)
|
||||
return newResourceClient(cc, cc)
|
||||
cci := grpchan.InterceptClientConn(indexChannel, clientInt.UnaryClientInterceptor, clientInt.StreamClientInterceptor)
|
||||
return newResourceClient(cc, cci)
|
||||
}
|
||||
|
||||
type RemoteResourceClientConfig struct {
|
||||
|
||||
@@ -127,7 +127,10 @@ type SearchBackend interface {
|
||||
GetOpenIndexes() []NamespacedResource
|
||||
}
|
||||
|
||||
// This supports indexing+search regardless of implementation
|
||||
var _ SearchServer = &searchSupport{}
|
||||
|
||||
// This supports indexing+search regardless of implementation.
|
||||
// Implements SearchServer interface.
|
||||
type searchSupport struct {
|
||||
log log.Logger
|
||||
storage StorageBackend
|
||||
@@ -160,6 +163,10 @@ var (
|
||||
_ 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) {
|
||||
// No backend search support
|
||||
if opts.Backend == nil {
|
||||
@@ -598,6 +605,22 @@ func (s *searchSupport) buildIndexes(ctx context.Context) (int, error) {
|
||||
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
|
||||
}
|
||||
|
||||
// IsHealthy implements resourcepb.DiagnosticsServer
|
||||
func (s *searchSupport) IsHealthy(ctx context.Context, req *resourcepb.HealthCheckRequest) (*resourcepb.HealthCheckResponse, error) {
|
||||
return &resourcepb.HealthCheckResponse{
|
||||
Status: resourcepb.HealthCheckResponse_SERVING,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *searchSupport) init(ctx context.Context) error {
|
||||
origCtx := ctx
|
||||
|
||||
@@ -863,7 +886,7 @@ func newRebuildRequest(key NamespacedResource, minBuildTime, lastImportTime time
|
||||
|
||||
func (s *searchSupport) getOrCreateIndex(ctx context.Context, stats *SearchStats, key NamespacedResource, reason string) (ResourceIndex, error) {
|
||||
if s == nil || s.search == nil {
|
||||
return nil, fmt.Errorf("search is not configured properly (missing unifiedStorageSearch feature toggle?)")
|
||||
return nil, fmt.Errorf("search is not configured properly (missing enable_search config?)")
|
||||
}
|
||||
|
||||
ctx, span := tracer.Start(ctx, "resource.searchSupport.getOrCreateIndex")
|
||||
|
||||
@@ -60,7 +60,7 @@ func ProvideSearchDistributorServer(cfg *setting.Cfg, features featuremgmt.Featu
|
||||
}
|
||||
|
||||
type RingClient struct {
|
||||
Client ResourceClient
|
||||
Client SearchClient
|
||||
grpc_health_v1.HealthClient
|
||||
Conn *grpc.ClientConn
|
||||
}
|
||||
@@ -99,7 +99,7 @@ var (
|
||||
func (ds *distributorServer) Search(ctx context.Context, r *resourcepb.ResourceSearchRequest) (*resourcepb.ResourceSearchResponse, error) {
|
||||
ctx, span := ds.tracing.Start(ctx, "distributor.Search")
|
||||
defer span.End()
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Options.Key.Namespace, "Search")
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Options.Key.Namespace)
|
||||
if err != nil {
|
||||
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) {
|
||||
ctx, span := ds.tracing.Start(ctx, "distributor.GetStats")
|
||||
defer span.End()
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace, "GetStats")
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace)
|
||||
if err != nil {
|
||||
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) {
|
||||
ctx, span := ds.tracing.Start(ctx, "distributor.CountManagedObjects")
|
||||
defer span.End()
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace, "CountManagedObjects")
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace)
|
||||
if err != nil {
|
||||
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) {
|
||||
ctx, span := ds.tracing.Start(ctx, "distributor.ListManagedObjects")
|
||||
defer span.End()
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace, "ListManagedObjects")
|
||||
ctx, client, err := ds.getClientToDistributeRequest(ctx, r.Namespace)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -234,7 +234,7 @@ func (ds *distributorServer) ListManagedObjects(ctx context.Context, r *resource
|
||||
return client.ListManagedObjects(ctx, r)
|
||||
}
|
||||
|
||||
func (ds *distributorServer) getClientToDistributeRequest(ctx context.Context, namespace string, methodName string) (context.Context, ResourceClient, error) {
|
||||
func (ds *distributorServer) getClientToDistributeRequest(ctx context.Context, namespace string) (context.Context, SearchClient, error) {
|
||||
ringHasher := fnv.New32a()
|
||||
_, err := ringHasher.Write([]byte(namespace))
|
||||
if err != nil {
|
||||
|
||||
@@ -34,12 +34,18 @@ import (
|
||||
|
||||
var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/resource")
|
||||
|
||||
type SearchServer interface {
|
||||
LifecycleHooks
|
||||
|
||||
resourcepb.ResourceIndexServer
|
||||
resourcepb.ManagedObjectIndexServer
|
||||
resourcepb.DiagnosticsServer
|
||||
}
|
||||
|
||||
// ResourceServer implements all gRPC services
|
||||
type ResourceServer interface {
|
||||
resourcepb.ResourceStoreServer
|
||||
resourcepb.BulkStoreServer
|
||||
resourcepb.ResourceIndexServer
|
||||
resourcepb.ManagedObjectIndexServer
|
||||
resourcepb.BlobStoreServer
|
||||
resourcepb.DiagnosticsServer
|
||||
resourcepb.QuotasServer
|
||||
@@ -221,9 +227,6 @@ type ResourceServerOptions struct {
|
||||
// The blob configuration
|
||||
Blob BlobConfig
|
||||
|
||||
// Search options
|
||||
Search SearchOptions
|
||||
|
||||
// Quota service
|
||||
OverridesService *OverridesService
|
||||
|
||||
@@ -251,16 +254,15 @@ type ResourceServerOptions struct {
|
||||
|
||||
storageMetrics *StorageMetrics
|
||||
|
||||
IndexMetrics *BleveIndexMetrics
|
||||
|
||||
// MaxPageSizeBytes is the maximum size of a page in bytes.
|
||||
MaxPageSizeBytes int
|
||||
// IndexMinUpdateInterval is the time to wait after a successful write operation to ensure read-after-write consistency in search.
|
||||
// This config is shared with search
|
||||
IndexMinUpdateInterval time.Duration
|
||||
|
||||
// QOSQueue is the quality of service queue used to enqueue
|
||||
QOSQueue QOSEnqueuer
|
||||
QOSConfig QueueConfig
|
||||
|
||||
OwnsIndexFn func(key NamespacedResource) (bool, error)
|
||||
}
|
||||
|
||||
func NewResourceServer(opts ResourceServerOptions) (*server, error) {
|
||||
@@ -343,23 +345,24 @@ func NewResourceServer(opts ResourceServerOptions) (*server, error) {
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
storageMetrics: opts.storageMetrics,
|
||||
indexMetrics: opts.IndexMetrics,
|
||||
maxPageSizeBytes: opts.MaxPageSizeBytes,
|
||||
reg: opts.Reg,
|
||||
queue: opts.QOSQueue,
|
||||
queueConfig: opts.QOSConfig,
|
||||
overridesService: opts.OverridesService,
|
||||
|
||||
artificialSuccessfulWriteDelay: opts.Search.IndexMinUpdateInterval,
|
||||
artificialSuccessfulWriteDelay: opts.IndexMinUpdateInterval,
|
||||
}
|
||||
|
||||
if opts.Search.Resources != nil {
|
||||
var err error
|
||||
s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.IndexMetrics, opts.OwnsIndexFn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
/*
|
||||
if opts.Search.Resources != nil {
|
||||
var err error
|
||||
s.search, err = newSearchSupport(opts.Search, s.backend, s.access, s.blob, opts.IndexMetrics, opts.OwnsIndexFn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
err := s.Init(ctx)
|
||||
if err != nil {
|
||||
@@ -377,7 +380,6 @@ type server struct {
|
||||
backend StorageBackend
|
||||
blob BlobSupport
|
||||
secure secrets.InlineSecureValueSupport
|
||||
search *searchSupport
|
||||
diagnostics resourcepb.DiagnosticsServer
|
||||
access claims.AccessClient
|
||||
writeHooks WriteAccessHooks
|
||||
@@ -424,11 +426,6 @@ func (s *server) Init(ctx context.Context) error {
|
||||
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
|
||||
if s.initErr == nil {
|
||||
s.initErr = s.initWatcher()
|
||||
@@ -453,10 +450,6 @@ func (s *server) Stop(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
|
||||
if s.search != nil {
|
||||
s.search.stop()
|
||||
}
|
||||
|
||||
if s.overridesService != nil {
|
||||
if err := s.overridesService.stop(ctx); err != nil {
|
||||
stopFailed = true
|
||||
@@ -1372,47 +1365,6 @@ 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.
|
||||
func (s *server) IsHealthy(ctx context.Context, req *resourcepb.HealthCheckRequest) (*resourcepb.HealthCheckResponse, error) {
|
||||
return s.diagnostics.IsHealthy(ctx, req)
|
||||
@@ -1568,14 +1520,6 @@ func (s *server) runInQueue(ctx context.Context, tenantID string, runnable func(
|
||||
}
|
||||
}
|
||||
|
||||
func (s *server) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) {
|
||||
if s.search == nil {
|
||||
return nil, fmt.Errorf("search index not configured")
|
||||
}
|
||||
|
||||
return s.search.RebuildIndexes(ctx, req)
|
||||
}
|
||||
|
||||
func (s *server) checkQuota(ctx context.Context, nsr NamespacedResource) {
|
||||
span := trace.SpanFromContext(ctx)
|
||||
span.AddEvent("checkQuota", trace.WithAttributes(
|
||||
|
||||
@@ -1253,21 +1253,23 @@ func (b *bleveIndex) toBleveSearchRequest(ctx context.Context, req *resourcepb.R
|
||||
queryExact.SetField(resource.SEARCH_FIELD_TITLE)
|
||||
queryExact.Analyzer = keyword.Name // don't analyze the query input - treat it as a single token
|
||||
queryExact.Operator = query.MatchQueryOperatorAnd // This doesn't make a difference for keyword analyzer, we add it just to be explicit.
|
||||
searchQuery := bleve.NewDisjunctionQuery(queryExact)
|
||||
|
||||
// Query 2: Phrase query with standard analyzer
|
||||
queryPhrase := bleve.NewMatchPhraseQuery(req.Query)
|
||||
queryPhrase.SetBoost(5.0)
|
||||
queryPhrase.SetField(resource.SEARCH_FIELD_TITLE)
|
||||
queryPhrase.Analyzer = standard.Name
|
||||
searchQuery.AddQuery(queryPhrase)
|
||||
|
||||
// Query 3: Match query with standard analyzer
|
||||
queryAnalyzed := bleve.NewMatchQuery(removeSmallTerms(req.Query))
|
||||
queryAnalyzed.SetField(resource.SEARCH_FIELD_TITLE)
|
||||
queryAnalyzed.SetBoost(2.0)
|
||||
queryAnalyzed.Analyzer = standard.Name
|
||||
queryAnalyzed.Operator = query.MatchQueryOperatorAnd // Make sure all terms from the query are matched
|
||||
searchQuery.AddQuery(queryAnalyzed)
|
||||
|
||||
// At least one of the queries must match
|
||||
searchQuery := bleve.NewDisjunctionQuery(queryExact, queryAnalyzed, queryPhrase)
|
||||
queries = append(queries, searchQuery)
|
||||
}
|
||||
|
||||
|
||||
@@ -23,7 +23,6 @@ import (
|
||||
"go.uber.org/goleak"
|
||||
|
||||
authlib "github.com/grafana/authlib/types"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/identity"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/utils"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
@@ -18,6 +19,7 @@ import (
|
||||
const DASHBOARD_SCHEMA_VERSION = "schema_version"
|
||||
const DASHBOARD_LINK_COUNT = "link_count"
|
||||
const DASHBOARD_PANEL_TYPES = "panel_types"
|
||||
const DASHBOARD_PANEL_TITLE = "panel_title"
|
||||
const DASHBOARD_DS_TYPES = "ds_types"
|
||||
const DASHBOARD_TRANSFORMATIONS = "transformation"
|
||||
const DASHBOARD_LIBRARY_PANEL_REFERENCE = "reference.LibraryPanel"
|
||||
@@ -53,11 +55,21 @@ func DashboardBuilder(namespaced resource.NamespacedDocumentSupplier) (resource.
|
||||
Type: resourcepb.ResourceTableColumnDefinition_INT32,
|
||||
Description: "How many links appear on the page",
|
||||
},
|
||||
{
|
||||
Name: DASHBOARD_PANEL_TITLE,
|
||||
Type: resourcepb.ResourceTableColumnDefinition_STRING,
|
||||
IsArray: true,
|
||||
Description: "The panel title text",
|
||||
Properties: &resourcepb.ResourceTableColumnDefinition_Properties{
|
||||
Filterable: false, // full text
|
||||
FreeText: true,
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: DASHBOARD_PANEL_TYPES,
|
||||
Type: resourcepb.ResourceTableColumnDefinition_STRING,
|
||||
IsArray: true,
|
||||
Description: "How many links appear on the page",
|
||||
Description: "The panel types used in this dashboard",
|
||||
Properties: &resourcepb.ResourceTableColumnDefinition_Properties{
|
||||
Filterable: true,
|
||||
},
|
||||
@@ -269,14 +281,22 @@ func (s *DashboardDocumentBuilder) BuildDocument(ctx context.Context, key *resou
|
||||
doc.Description = summary.Description
|
||||
doc.Tags = summary.Tags
|
||||
|
||||
panelTitles := []string{}
|
||||
panelTypes := []string{}
|
||||
transformations := []string{}
|
||||
dsTypes := []string{}
|
||||
|
||||
for _, p := range summary.Panels {
|
||||
if p.Type != "" {
|
||||
for p := range summary.PanelIterator() {
|
||||
switch p.Type {
|
||||
case "": // ignore
|
||||
case "row": // row should map to a layout type when we support v2 constructs
|
||||
default:
|
||||
panelTypes = append(panelTypes, p.Type)
|
||||
}
|
||||
|
||||
if len(p.Title) > 0 {
|
||||
panelTitles = append(panelTitles, p.Title)
|
||||
}
|
||||
if len(p.Transformer) > 0 {
|
||||
transformations = append(transformations, p.Transformer...)
|
||||
}
|
||||
@@ -309,17 +329,20 @@ func (s *DashboardDocumentBuilder) BuildDocument(ctx context.Context, key *resou
|
||||
resource.SEARCH_FIELD_LEGACY_ID: summary.ID,
|
||||
}
|
||||
|
||||
if len(panelTitles) > 0 {
|
||||
doc.Fields[DASHBOARD_PANEL_TITLE] = panelTitles
|
||||
}
|
||||
if len(panelTypes) > 0 {
|
||||
sort.Strings(panelTypes)
|
||||
doc.Fields[DASHBOARD_PANEL_TYPES] = panelTypes
|
||||
doc.Fields[DASHBOARD_PANEL_TYPES] = slices.Compact(panelTypes) // distinct values
|
||||
}
|
||||
if len(dsTypes) > 0 {
|
||||
sort.Strings(dsTypes)
|
||||
doc.Fields[DASHBOARD_DS_TYPES] = dsTypes
|
||||
doc.Fields[DASHBOARD_DS_TYPES] = slices.Compact(dsTypes) // distinct values
|
||||
}
|
||||
if len(transformations) > 0 {
|
||||
sort.Strings(transformations)
|
||||
doc.Fields[DASHBOARD_TRANSFORMATIONS] = transformations
|
||||
doc.Fields[DASHBOARD_TRANSFORMATIONS] = slices.Compact(transformations) // distinct values
|
||||
}
|
||||
|
||||
for k, v := range s.Stats[summary.UID] {
|
||||
|
||||
@@ -32,10 +32,16 @@
|
||||
"errors_last_7_days": 1,
|
||||
"grafana.app/deprecatedInternalID": 141,
|
||||
"link_count": 0,
|
||||
"panel_title": [
|
||||
"green pie",
|
||||
"red pie",
|
||||
"blue pie",
|
||||
"collapsed row"
|
||||
],
|
||||
"panel_types": [
|
||||
"barchart",
|
||||
"graph",
|
||||
"row"
|
||||
"pie"
|
||||
],
|
||||
"schema_version": 38
|
||||
},
|
||||
@@ -46,6 +52,12 @@
|
||||
"kind": "DataSource",
|
||||
"name": "DSUID"
|
||||
},
|
||||
{
|
||||
"relation": "depends-on",
|
||||
"group": "dashboards.grafana.app",
|
||||
"kind": "LibraryPanel",
|
||||
"name": "l3d2s634-fdgf-75u4-3fg3-67j966ii7jur"
|
||||
},
|
||||
{
|
||||
"relation": "depends-on",
|
||||
"group": "dashboards.grafana.app",
|
||||
|
||||
@@ -67,7 +67,7 @@
|
||||
"name": "red pie",
|
||||
"uid": "e1d5f519-dabd-47c6-9ad7-83d181ce1cee"
|
||||
},
|
||||
"title": "green pie"
|
||||
"title": "red pie"
|
||||
},
|
||||
{
|
||||
"id": 7,
|
||||
@@ -78,6 +78,14 @@
|
||||
"id": 8,
|
||||
"type": "graph"
|
||||
},
|
||||
{
|
||||
"id": 20,
|
||||
"type": "graph"
|
||||
},
|
||||
{
|
||||
"id": 30,
|
||||
"type": "graph"
|
||||
},
|
||||
{
|
||||
"collapsed": true,
|
||||
"gridPos": {
|
||||
@@ -101,6 +109,10 @@
|
||||
"uid": "l3d2s634-fdgf-75u4-3fg3-67j966ii7jur"
|
||||
},
|
||||
"title": "blue pie"
|
||||
},
|
||||
{
|
||||
"id": 40,
|
||||
"type": "pie"
|
||||
}
|
||||
],
|
||||
"title": "collapsed row",
|
||||
|
||||
@@ -19,7 +19,7 @@ func NewSearchOptions(
|
||||
ownsIndexFn func(key resource.NamespacedResource) (bool, error),
|
||||
) (resource.SearchOptions, error) {
|
||||
//nolint:staticcheck // not yet migrated to OpenFeature
|
||||
if cfg.EnableSearch || features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageSearch) || features.IsEnabledGlobally(featuremgmt.FlagProvisioning) {
|
||||
if cfg.EnableSearch || features.IsEnabledGlobally(featuremgmt.FlagProvisioning) {
|
||||
root := cfg.IndexPath
|
||||
if root == "" {
|
||||
root = filepath.Join(cfg.DataPath, "unified-search", "bleve")
|
||||
|
||||
+11
-1
@@ -71,11 +71,18 @@
|
||||
"description": "How many links appear on the page",
|
||||
"priority": 0
|
||||
},
|
||||
{
|
||||
"name": "panel_title",
|
||||
"type": "string",
|
||||
"format": "",
|
||||
"description": "The panel title text",
|
||||
"priority": 0
|
||||
},
|
||||
{
|
||||
"name": "panel_types",
|
||||
"type": "string",
|
||||
"format": "",
|
||||
"description": "How many links appear on the page",
|
||||
"description": "The panel types used in this dashboard",
|
||||
"priority": 0
|
||||
},
|
||||
{
|
||||
@@ -214,6 +221,7 @@
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null
|
||||
],
|
||||
"object": {
|
||||
@@ -239,6 +247,7 @@
|
||||
"repo",
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
[
|
||||
"timeseries"
|
||||
],
|
||||
@@ -282,6 +291,7 @@
|
||||
"repo",
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
[
|
||||
"timeseries",
|
||||
"table"
|
||||
|
||||
@@ -11,6 +11,9 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-sql-driver/mysql"
|
||||
infraDB "github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/services/sqlstore/migrator"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/lib/pq"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
@@ -44,10 +47,63 @@ const defaultPrunerHistoryLimit = 20
|
||||
|
||||
func ProvideStorageBackend(
|
||||
cfg *setting.Cfg,
|
||||
db infraDB.DB,
|
||||
tracer trace.Tracer,
|
||||
reg prometheus.Registerer,
|
||||
storageMetrics *resource.StorageMetrics,
|
||||
) (resource.StorageBackend, error) {
|
||||
// TODO: make this the central place to provide SQL backend
|
||||
// Currently it is skipped as we need to handle the cases of Diagnostics and Lifecycle
|
||||
return nil, nil
|
||||
// Create the resource DB
|
||||
eDB, err := dbimpl.ProvideResourceDB(db, cfg, tracer)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create resource DB: %w", err)
|
||||
}
|
||||
|
||||
// Check if HA is enabled
|
||||
isHA := isHighAvailabilityEnabled(
|
||||
cfg.SectionWithEnvOverrides("database"),
|
||||
cfg.SectionWithEnvOverrides("resource_api"),
|
||||
)
|
||||
|
||||
// Create the backend
|
||||
backend, err := NewBackend(BackendOptions{
|
||||
DBProvider: eDB,
|
||||
Reg: reg,
|
||||
IsHA: isHA,
|
||||
storageMetrics: storageMetrics,
|
||||
LastImportTimeMaxAge: cfg.MaxFileIndexAge,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create backend: %w", err)
|
||||
}
|
||||
|
||||
// Initialize the backend
|
||||
if err := backend.Init(context.Background()); err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize backend: %w", err)
|
||||
}
|
||||
|
||||
return backend, nil
|
||||
}
|
||||
|
||||
// isHighAvailabilityEnabled determines if high availability mode should
|
||||
// be enabled based on database configuration. High availability is enabled
|
||||
// by default except for SQLite databases.
|
||||
func isHighAvailabilityEnabled(dbCfg, resourceAPICfg *setting.DynamicSection) bool {
|
||||
// If the resource API is using a non-SQLite database, we assume it's in HA mode.
|
||||
resourceDBType := resourceAPICfg.Key("db_type").String()
|
||||
if resourceDBType != "" && resourceDBType != migrator.SQLite {
|
||||
return true
|
||||
}
|
||||
|
||||
// Check in the config if HA is enabled - by default we always assume a HA setup.
|
||||
isHA := dbCfg.Key("high_availability").MustBool(true)
|
||||
|
||||
// SQLite is not possible to run in HA, so we force it to false.
|
||||
databaseType := dbCfg.Key("type").String()
|
||||
if databaseType == migrator.SQLite {
|
||||
isHA = false
|
||||
}
|
||||
|
||||
return isHA
|
||||
}
|
||||
|
||||
type Backend interface {
|
||||
|
||||
@@ -0,0 +1,322 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/search"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"google.golang.org/grpc/health/grpc_health_v1"
|
||||
|
||||
"github.com/grafana/dskit/kv"
|
||||
"github.com/grafana/dskit/ring"
|
||||
"github.com/grafana/dskit/services"
|
||||
|
||||
infraDB "github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/modules"
|
||||
"github.com/grafana/grafana/pkg/services/authz"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/grpcserver"
|
||||
"github.com/grafana/grafana/pkg/services/grpcserver/interceptors"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource/grpc"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||
)
|
||||
|
||||
var _ UnifiedStorageGrpcService = (*searchService)(nil)
|
||||
|
||||
// operation used by the search-servers to check if they own the namespace
|
||||
var (
|
||||
searchOwnerRead = ring.NewOp([]ring.InstanceState{ring.JOINING, ring.ACTIVE, ring.LEAVING}, nil)
|
||||
)
|
||||
|
||||
type searchService struct {
|
||||
*services.BasicService
|
||||
|
||||
backend resource.StorageBackend
|
||||
cfg *setting.Cfg
|
||||
features featuremgmt.FeatureToggles
|
||||
db infraDB.DB
|
||||
stopCh chan struct{}
|
||||
stoppedCh chan error
|
||||
handler grpcserver.Provider
|
||||
tracing trace.Tracer
|
||||
authenticator func(ctx context.Context) (context.Context, error)
|
||||
httpServerRouter *mux.Router
|
||||
|
||||
log log.Logger
|
||||
reg prometheus.Registerer
|
||||
docBuilders resource.DocumentBuilderSupplier
|
||||
indexMetrics *resource.BleveIndexMetrics
|
||||
searchRing *ring.Ring
|
||||
|
||||
// Ring lifecycle and sharding support
|
||||
ringLifecycler *ring.BasicLifecycler
|
||||
|
||||
// Subservices manager
|
||||
subservices *services.Manager
|
||||
subservicesWatcher *services.FailureWatcher
|
||||
hasSubservices bool
|
||||
}
|
||||
|
||||
func ProvideUnifiedSearchGrpcService(
|
||||
cfg *setting.Cfg,
|
||||
features featuremgmt.FeatureToggles,
|
||||
db infraDB.DB,
|
||||
log log.Logger,
|
||||
reg prometheus.Registerer,
|
||||
docBuilders resource.DocumentBuilderSupplier,
|
||||
indexMetrics *resource.BleveIndexMetrics,
|
||||
searchRing *ring.Ring,
|
||||
memberlistKVConfig kv.Config,
|
||||
backend resource.StorageBackend,
|
||||
httpServerRouter *mux.Router,
|
||||
) (UnifiedStorageGrpcService, error) {
|
||||
var err error
|
||||
tracer := otel.Tracer("unified-search-server")
|
||||
|
||||
authn := NewAuthenticatorWithFallback(cfg, reg, tracer, func(ctx context.Context) (context.Context, error) {
|
||||
auth := grpc.Authenticator{Tracer: tracer}
|
||||
return auth.Authenticate(ctx)
|
||||
})
|
||||
|
||||
s := &searchService{
|
||||
backend: backend,
|
||||
cfg: cfg,
|
||||
features: features,
|
||||
stopCh: make(chan struct{}),
|
||||
stoppedCh: make(chan error, 1),
|
||||
authenticator: authn,
|
||||
tracing: tracer,
|
||||
db: db,
|
||||
log: log,
|
||||
reg: reg,
|
||||
docBuilders: docBuilders,
|
||||
indexMetrics: indexMetrics,
|
||||
searchRing: searchRing,
|
||||
httpServerRouter: httpServerRouter,
|
||||
subservicesWatcher: services.NewFailureWatcher(),
|
||||
}
|
||||
|
||||
subservices := []services.Service{}
|
||||
if cfg.EnableSharding {
|
||||
ringStore, err := kv.NewClient(
|
||||
memberlistKVConfig,
|
||||
ring.GetCodec(),
|
||||
kv.RegistererWithKVName(reg, resource.RingName),
|
||||
log,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create KV store client: %s", err)
|
||||
}
|
||||
|
||||
lifecyclerCfg, err := toLifecyclerConfig(cfg, log)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize search-ring lifecycler config: %s", err)
|
||||
}
|
||||
|
||||
// Define lifecycler delegates in reverse order (last to be called defined first because they're
|
||||
// chained via "next delegate").
|
||||
delegate := ring.BasicLifecyclerDelegate(ring.NewInstanceRegisterDelegate(ring.JOINING, resource.RingNumTokens))
|
||||
delegate = ring.NewLeaveOnStoppingDelegate(delegate, log)
|
||||
delegate = ring.NewAutoForgetDelegate(resource.RingHeartbeatTimeout*2, delegate, log)
|
||||
|
||||
s.ringLifecycler, err = ring.NewBasicLifecycler(
|
||||
lifecyclerCfg,
|
||||
resource.RingName,
|
||||
resource.RingKey,
|
||||
ringStore,
|
||||
delegate,
|
||||
log,
|
||||
reg,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize search-ring lifecycler: %s", err)
|
||||
}
|
||||
|
||||
s.ringLifecycler.SetKeepInstanceInTheRingOnShutdown(true)
|
||||
subservices = append(subservices, s.ringLifecycler)
|
||||
}
|
||||
|
||||
if len(subservices) > 0 {
|
||||
s.hasSubservices = true
|
||||
s.subservices, err = services.NewManager(subservices...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create subservices manager: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// This will be used when running as a dskit service
|
||||
s.BasicService = services.NewBasicService(s.starting, s.running, s.stopping).WithName(modules.SearchServer)
|
||||
|
||||
// Register HTTP endpoints if router is provided
|
||||
s.RegisterHTTPEndpoints(httpServerRouter)
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *searchService) PrepareDownscale(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.Method {
|
||||
case http.MethodPost:
|
||||
s.log.Info("Preparing for downscale. Will not keep instance in ring on shutdown.")
|
||||
s.ringLifecycler.SetKeepInstanceInTheRingOnShutdown(false)
|
||||
case http.MethodDelete:
|
||||
s.log.Info("Downscale canceled. Will keep instance in ring on shutdown.")
|
||||
s.ringLifecycler.SetKeepInstanceInTheRingOnShutdown(true)
|
||||
case http.MethodGet:
|
||||
// used for delayed downscale use case, which we don't support. Leaving here for completion sake
|
||||
s.log.Info("Received GET request for prepare-downscale. Behavior not implemented.")
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (s *searchService) OwnsIndex(key resource.NamespacedResource) (bool, error) {
|
||||
if s.searchRing == nil {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
if st := s.searchRing.State(); st != services.Running {
|
||||
return false, fmt.Errorf("ring is not Running: %s", st)
|
||||
}
|
||||
|
||||
ringHasher := fnv.New32a()
|
||||
_, err := ringHasher.Write([]byte(key.Namespace))
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("error hashing namespace: %w", err)
|
||||
}
|
||||
|
||||
rs, err := s.searchRing.GetWithOptions(ringHasher.Sum32(), searchOwnerRead, ring.WithReplicationFactor(s.searchRing.ReplicationFactor()))
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("error getting replicaset from ring: %w", err)
|
||||
}
|
||||
|
||||
return rs.Includes(s.ringLifecycler.GetInstanceAddr()), nil
|
||||
}
|
||||
|
||||
func (s *searchService) starting(ctx context.Context) error {
|
||||
if s.hasSubservices {
|
||||
s.subservicesWatcher.WatchManager(s.subservices)
|
||||
if err := services.StartManagerAndAwaitHealthy(ctx, s.subservices); err != nil {
|
||||
return fmt.Errorf("failed to start subservices: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
authzClient, err := authz.ProvideStandaloneAuthZClient(s.cfg, s.features, s.tracing, s.reg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create search options for the search server
|
||||
searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create the search server
|
||||
searchServer, err := NewSearchServer(SearchServerOptions{
|
||||
Backend: s.backend,
|
||||
DB: s.db,
|
||||
Cfg: s.cfg,
|
||||
Tracer: s.tracing,
|
||||
Reg: s.reg,
|
||||
AccessClient: authzClient,
|
||||
SearchOptions: searchOptions,
|
||||
IndexMetrics: s.indexMetrics,
|
||||
OwnsIndexFn: s.OwnsIndex,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.handler, err = grpcserver.ProvideService(s.cfg, s.features, interceptors.AuthenticatorFunc(s.authenticator), s.tracing, prometheus.DefaultRegisterer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
healthService, err := resource.ProvideHealthService(searchServer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
srv := s.handler.GetServer()
|
||||
// Register search services
|
||||
resourcepb.RegisterResourceIndexServer(srv, searchServer)
|
||||
resourcepb.RegisterManagedObjectIndexServer(srv, searchServer)
|
||||
resourcepb.RegisterDiagnosticsServer(srv, searchServer)
|
||||
grpc_health_v1.RegisterHealthServer(srv, healthService)
|
||||
|
||||
// register reflection service
|
||||
_, err = grpcserver.ProvideReflectionService(s.cfg, s.handler)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if s.cfg.EnableSharding {
|
||||
s.log.Info("waiting until search server is JOINING in the ring")
|
||||
lfcCtx, cancel := context.WithTimeout(context.Background(), s.cfg.ResourceServerJoinRingTimeout)
|
||||
defer cancel()
|
||||
if err := ring.WaitInstanceState(lfcCtx, s.searchRing, s.ringLifecycler.GetInstanceID(), ring.JOINING); err != nil {
|
||||
return fmt.Errorf("error switching to JOINING in the ring: %s", err)
|
||||
}
|
||||
s.log.Info("search server is JOINING in the ring")
|
||||
|
||||
if err := s.ringLifecycler.ChangeState(ctx, ring.ACTIVE); err != nil {
|
||||
return fmt.Errorf("error switching to ACTIVE in the ring: %s", err)
|
||||
}
|
||||
s.log.Info("search server is ACTIVE in the ring")
|
||||
}
|
||||
|
||||
// start the gRPC server
|
||||
go func() {
|
||||
err := s.handler.Run(ctx)
|
||||
if err != nil {
|
||||
s.stoppedCh <- err
|
||||
} else {
|
||||
s.stoppedCh <- nil
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetAddress returns the address of the gRPC server.
|
||||
func (s *searchService) GetAddress() string {
|
||||
return s.handler.GetAddress()
|
||||
}
|
||||
|
||||
func (s *searchService) running(ctx context.Context) error {
|
||||
select {
|
||||
case err := <-s.stoppedCh:
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
return err
|
||||
}
|
||||
case err := <-s.subservicesWatcher.Chan():
|
||||
return fmt.Errorf("subservice failure: %w", err)
|
||||
case <-ctx.Done():
|
||||
close(s.stopCh)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *searchService) stopping(_ error) error {
|
||||
if s.hasSubservices {
|
||||
err := services.StopManagerAndAwaitStopped(context.Background(), s.subservices)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to stop subservices: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *searchService) RegisterHTTPEndpoints(httpServerRouter *mux.Router) {
|
||||
if httpServerRouter != nil && s.cfg.EnableSharding {
|
||||
httpServerRouter.Path("/prepare-downscale").Methods("GET", "POST", "DELETE").Handler(http.HandlerFunc(s.PrepareDownscale))
|
||||
}
|
||||
}
|
||||
@@ -16,7 +16,6 @@ import (
|
||||
secrets "github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
|
||||
inlinesecurevalue "github.com/grafana/grafana/pkg/registry/apis/secret/inline"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/sqlstore/migrator"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl"
|
||||
@@ -30,8 +29,21 @@ type QOSEnqueueDequeuer interface {
|
||||
Dequeue(ctx context.Context) (func(), error)
|
||||
}
|
||||
|
||||
// ServerOptions contains the options for creating a new ResourceServer
|
||||
type ServerOptions struct {
|
||||
// SearchServerOptions contains the options for creating a new SearchServer
|
||||
type SearchServerOptions struct {
|
||||
Backend resource.StorageBackend
|
||||
DB infraDB.DB
|
||||
Cfg *setting.Cfg
|
||||
Tracer trace.Tracer
|
||||
Reg prometheus.Registerer
|
||||
AccessClient types.AccessClient
|
||||
SearchOptions resource.SearchOptions
|
||||
IndexMetrics *resource.BleveIndexMetrics
|
||||
OwnsIndexFn func(key resource.NamespacedResource) (bool, error)
|
||||
}
|
||||
|
||||
// StorageServerOptions contains the options for creating a storage-only server (without search)
|
||||
type StorageServerOptions struct {
|
||||
Backend resource.StorageBackend
|
||||
OverridesService *resource.OverridesService
|
||||
DB infraDB.DB
|
||||
@@ -39,16 +51,66 @@ type ServerOptions struct {
|
||||
Tracer trace.Tracer
|
||||
Reg prometheus.Registerer
|
||||
AccessClient types.AccessClient
|
||||
SearchOptions resource.SearchOptions
|
||||
StorageMetrics *resource.StorageMetrics
|
||||
IndexMetrics *resource.BleveIndexMetrics
|
||||
Features featuremgmt.FeatureToggles
|
||||
QOSQueue QOSEnqueueDequeuer
|
||||
SecureValues secrets.InlineSecureValueSupport
|
||||
OwnsIndexFn func(key resource.NamespacedResource) (bool, error)
|
||||
}
|
||||
|
||||
func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
|
||||
// NewSearchServer creates a new SearchServer with the given options.
|
||||
// This can be used to create a standalone search server or to create a search server
|
||||
// that will be passed to NewResourceServer.
|
||||
//
|
||||
// Important: When running in monolith mode, the backend should be provided by the caller
|
||||
// to avoid duplicate metrics registration. Only in standalone microservice mode should
|
||||
// this function create its own backend.
|
||||
func NewSearchServer(opts SearchServerOptions) (resource.SearchServer, error) {
|
||||
backend := opts.Backend
|
||||
if backend == nil {
|
||||
eDB, err := dbimpl.ProvideResourceDB(opts.DB, opts.Cfg, opts.Tracer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
isHA := isHighAvailabilityEnabled(opts.Cfg.SectionWithEnvOverrides("database"),
|
||||
opts.Cfg.SectionWithEnvOverrides("resource_api"))
|
||||
|
||||
b, err := NewBackend(BackendOptions{
|
||||
DBProvider: eDB,
|
||||
Reg: opts.Reg,
|
||||
IsHA: isHA,
|
||||
LastImportTimeMaxAge: opts.SearchOptions.MaxIndexAge,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Initialize the backend before creating search server
|
||||
if err := b.Init(context.Background()); err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize backend: %w", err)
|
||||
}
|
||||
backend = b
|
||||
}
|
||||
|
||||
search, err := resource.NewSearchServer(opts.SearchOptions, backend, opts.AccessClient, nil, opts.IndexMetrics, opts.OwnsIndexFn)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create search server: %w", err)
|
||||
}
|
||||
|
||||
if err := search.Init(context.Background()); err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize search server: %w", err)
|
||||
}
|
||||
|
||||
return search, nil
|
||||
}
|
||||
|
||||
// NewStorageServer creates a storage-only server without search capabilities.
|
||||
// Use this when you want to run storage and search as separate services.
|
||||
//
|
||||
// Important: When running in monolith mode, the backend should be provided by the caller
|
||||
// to avoid duplicate metrics registration. Only in standalone microservice mode should
|
||||
// this function create its own backend.
|
||||
func NewStorageServer(opts StorageServerOptions) (resource.ResourceServer, error) {
|
||||
apiserverCfg := opts.Cfg.SectionWithEnvOverrides("grafana-apiserver")
|
||||
|
||||
if opts.SecureValues == nil && opts.Cfg != nil && opts.Cfg.SecretsManagement.GrpcClientEnable {
|
||||
@@ -92,7 +154,6 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
|
||||
|
||||
if opts.Backend != nil {
|
||||
serverOptions.Backend = opts.Backend
|
||||
// TODO: we should probably have a proper interface for diagnostics/lifecycle
|
||||
} else {
|
||||
eDB, err := dbimpl.ProvideResourceDB(opts.DB, opts.Cfg, opts.Tracer)
|
||||
if err != nil {
|
||||
@@ -130,7 +191,6 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
|
||||
return nil, fmt.Errorf("failed to create resource version manager: %w", err)
|
||||
}
|
||||
|
||||
// TODO add config to decide whether to pass RvManager or not
|
||||
kvBackendOpts.RvManager = rvManager
|
||||
kvBackend, err := resource.NewKVStorageBackend(kvBackendOpts)
|
||||
if err != nil {
|
||||
@@ -148,7 +208,7 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
|
||||
Reg: opts.Reg,
|
||||
IsHA: isHA,
|
||||
storageMetrics: opts.StorageMetrics,
|
||||
LastImportTimeMaxAge: opts.SearchOptions.MaxIndexAge, // No need to keep last_import_times older than max index age.
|
||||
LastImportTimeMaxAge: opts.Cfg.MaxFileIndexAge,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -159,33 +219,15 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
|
||||
}
|
||||
}
|
||||
|
||||
serverOptions.Search = opts.SearchOptions
|
||||
serverOptions.IndexMetrics = opts.IndexMetrics
|
||||
// Initialize the backend before creating server
|
||||
if serverOptions.Lifecycle != nil {
|
||||
if err := serverOptions.Lifecycle.Init(context.Background()); err != nil {
|
||||
return nil, fmt.Errorf("failed to initialize backend: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
serverOptions.QOSQueue = opts.QOSQueue
|
||||
serverOptions.OwnsIndexFn = opts.OwnsIndexFn
|
||||
serverOptions.OverridesService = opts.OverridesService
|
||||
|
||||
return resource.NewResourceServer(serverOptions)
|
||||
}
|
||||
|
||||
// isHighAvailabilityEnabled determines if high availability mode should
|
||||
// be enabled based on database configuration. High availability is enabled
|
||||
// by default except for SQLite databases.
|
||||
func isHighAvailabilityEnabled(dbCfg, resourceAPICfg *setting.DynamicSection) bool {
|
||||
// If the resource API is using a non-SQLite database, we assume it's in HA mode.
|
||||
resourceDBType := resourceAPICfg.Key("db_type").String()
|
||||
if resourceDBType != "" && resourceDBType != migrator.SQLite {
|
||||
return true
|
||||
}
|
||||
|
||||
// Check in the config if HA is enabled - by default we always assume a HA setup.
|
||||
isHA := dbCfg.Key("high_availability").MustBool(true)
|
||||
|
||||
// SQLite is not possible to run in HA, so we force it to false.
|
||||
databaseType := dbCfg.Key("type").String()
|
||||
if databaseType == migrator.SQLite {
|
||||
isHA = false
|
||||
}
|
||||
|
||||
return isHA
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/search"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promauto"
|
||||
"go.opentelemetry.io/otel"
|
||||
@@ -36,7 +37,7 @@ import (
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource/grpc"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/search"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl"
|
||||
"github.com/grafana/grafana/pkg/util/scheduler"
|
||||
)
|
||||
|
||||
@@ -54,38 +55,40 @@ type UnifiedStorageGrpcService interface {
|
||||
type service struct {
|
||||
*services.BasicService
|
||||
|
||||
// Subservices manager
|
||||
subservices *services.Manager
|
||||
subservicesWatcher *services.FailureWatcher
|
||||
hasSubservices bool
|
||||
|
||||
backend resource.StorageBackend
|
||||
cfg *setting.Cfg
|
||||
features featuremgmt.FeatureToggles
|
||||
db infraDB.DB
|
||||
stopCh chan struct{}
|
||||
stoppedCh chan error
|
||||
|
||||
handler grpcserver.Provider
|
||||
|
||||
tracing trace.Tracer
|
||||
|
||||
authenticator func(ctx context.Context) (context.Context, error)
|
||||
|
||||
backend resource.StorageBackend
|
||||
cfg *setting.Cfg
|
||||
features featuremgmt.FeatureToggles
|
||||
stopCh chan struct{}
|
||||
stoppedCh chan error
|
||||
authenticator func(context.Context) (context.Context, error)
|
||||
tracing trace.Tracer
|
||||
db infraDB.DB
|
||||
log log.Logger
|
||||
reg prometheus.Registerer
|
||||
docBuilders resource.DocumentBuilderSupplier
|
||||
storageMetrics *resource.StorageMetrics
|
||||
indexMetrics *resource.BleveIndexMetrics
|
||||
|
||||
docBuilders resource.DocumentBuilderSupplier
|
||||
|
||||
searchRing *ring.Ring
|
||||
|
||||
// Handler for the gRPC server
|
||||
handler grpcserver.Provider
|
||||
|
||||
// Ring lifecycle and sharding support
|
||||
ringLifecycler *ring.BasicLifecycler
|
||||
|
||||
queue QOSEnqueueDequeuer
|
||||
// QoS support
|
||||
queue *scheduler.Queue
|
||||
scheduler *scheduler.Scheduler
|
||||
|
||||
// Subservices management
|
||||
hasSubservices bool
|
||||
subservices *services.Manager
|
||||
subservicesWatcher *services.FailureWatcher
|
||||
}
|
||||
|
||||
// ProvideUnifiedStorageGrpcService provides a combined storage and search service running on the same gRPC server.
|
||||
// This is used when running Grafana as a monolith where both services share the same process.
|
||||
// Each service (storage and search) maintains its own lifecycle but shares the gRPC server.
|
||||
func ProvideUnifiedStorageGrpcService(
|
||||
cfg *setting.Cfg,
|
||||
features featuremgmt.FeatureToggles,
|
||||
@@ -101,7 +104,7 @@ func ProvideUnifiedStorageGrpcService(
|
||||
backend resource.StorageBackend,
|
||||
) (UnifiedStorageGrpcService, error) {
|
||||
var err error
|
||||
tracer := otel.Tracer("unified-storage")
|
||||
tracer := otel.Tracer("unified-storage-combined")
|
||||
|
||||
// FIXME: This is a temporary solution while we are migrating to the new authn interceptor
|
||||
// grpcutils.NewGrpcAuthenticator should be used instead.
|
||||
@@ -178,7 +181,7 @@ func ProvideUnifiedStorageGrpcService(
|
||||
MaxSizePerTenant: cfg.QOSMaxSizePerTenant,
|
||||
Registerer: qosReg,
|
||||
})
|
||||
scheduler, err := scheduler.NewScheduler(queue, &scheduler.Config{
|
||||
sched, err := scheduler.NewScheduler(queue, &scheduler.Config{
|
||||
NumWorkers: cfg.QOSNumberWorker,
|
||||
Logger: log,
|
||||
})
|
||||
@@ -187,7 +190,7 @@ func ProvideUnifiedStorageGrpcService(
|
||||
}
|
||||
|
||||
s.queue = queue
|
||||
s.scheduler = scheduler
|
||||
s.scheduler = sched
|
||||
subservices = append(subservices, s.queue, s.scheduler)
|
||||
}
|
||||
|
||||
@@ -200,6 +203,7 @@ func ProvideUnifiedStorageGrpcService(
|
||||
}
|
||||
|
||||
// This will be used when running as a dskit service
|
||||
// Note: We use StorageServer as the module name for backward compatibility
|
||||
s.BasicService = services.NewBasicService(s.starting, s.running, s.stopping).WithName(modules.StorageServer)
|
||||
|
||||
return s, nil
|
||||
@@ -220,11 +224,6 @@ func (s *service) PrepareDownscale(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
var (
|
||||
// operation used by the search-servers to check if they own the namespace
|
||||
searchOwnerRead = ring.NewOp([]ring.InstanceState{ring.JOINING, ring.ACTIVE, ring.LEAVING}, nil)
|
||||
)
|
||||
|
||||
func (s *service) OwnsIndex(key resource.NamespacedResource) (bool, error) {
|
||||
if s.searchRing == nil {
|
||||
return true, nil
|
||||
@@ -261,59 +260,108 @@ func (s *service) starting(ctx context.Context) error {
|
||||
return err
|
||||
}
|
||||
|
||||
searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
serverOptions := ServerOptions{
|
||||
Backend: s.backend,
|
||||
DB: s.db,
|
||||
Cfg: s.cfg,
|
||||
Tracer: s.tracing,
|
||||
Reg: s.reg,
|
||||
AccessClient: authzClient,
|
||||
SearchOptions: searchOptions,
|
||||
StorageMetrics: s.storageMetrics,
|
||||
IndexMetrics: s.indexMetrics,
|
||||
Features: s.features,
|
||||
QOSQueue: s.queue,
|
||||
OwnsIndexFn: s.OwnsIndex,
|
||||
}
|
||||
|
||||
// Setup overrides service if enabled
|
||||
var overridesSvc *resource.OverridesService
|
||||
if s.cfg.OverridesFilePath != "" {
|
||||
overridesSvc, err := resource.NewOverridesService(context.Background(), s.log, s.reg, s.tracing, resource.ReloadOptions{
|
||||
overridesSvc, err = resource.NewOverridesService(context.Background(), s.log, s.reg, s.tracing, resource.ReloadOptions{
|
||||
FilePath: s.cfg.OverridesFilePath,
|
||||
ReloadPeriod: s.cfg.OverridesReloadInterval,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
serverOptions.OverridesService = overridesSvc
|
||||
}
|
||||
|
||||
server, err := NewResourceServer(serverOptions)
|
||||
// Ensure we have a backend - create one if needed
|
||||
// This is critical: we create the backend ONCE and share it between search and storage servers
|
||||
// to avoid duplicate metrics registration
|
||||
backend := s.backend
|
||||
if backend == nil {
|
||||
eDB, err := dbimpl.ProvideResourceDB(s.db, s.cfg, s.tracing)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create resource DB: %w", err)
|
||||
}
|
||||
|
||||
isHA := isHighAvailabilityEnabled(s.cfg.SectionWithEnvOverrides("database"),
|
||||
s.cfg.SectionWithEnvOverrides("resource_api"))
|
||||
|
||||
b, err := NewBackend(BackendOptions{
|
||||
DBProvider: eDB,
|
||||
Reg: s.reg,
|
||||
IsHA: isHA,
|
||||
storageMetrics: s.storageMetrics,
|
||||
LastImportTimeMaxAge: s.cfg.MaxFileIndexAge,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create backend: %w", err)
|
||||
}
|
||||
|
||||
// Initialize the backend
|
||||
if err := b.Init(context.Background()); err != nil {
|
||||
return fmt.Errorf("failed to initialize backend: %w", err)
|
||||
}
|
||||
backend = b
|
||||
}
|
||||
|
||||
// Create search options for the search server
|
||||
searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create the search server - pass the shared backend
|
||||
searchServer, err := NewSearchServer(SearchServerOptions{
|
||||
Backend: backend, // Use the shared backend
|
||||
DB: s.db,
|
||||
Cfg: s.cfg,
|
||||
Tracer: s.tracing,
|
||||
Reg: s.reg,
|
||||
AccessClient: authzClient,
|
||||
SearchOptions: searchOptions,
|
||||
IndexMetrics: s.indexMetrics,
|
||||
OwnsIndexFn: s.OwnsIndex,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create the storage server - pass the shared backend
|
||||
storageServer, err := NewStorageServer(StorageServerOptions{
|
||||
Backend: backend, // Use the shared backend
|
||||
OverridesService: overridesSvc,
|
||||
DB: s.db,
|
||||
Cfg: s.cfg,
|
||||
Tracer: s.tracing,
|
||||
Reg: s.reg,
|
||||
AccessClient: authzClient,
|
||||
StorageMetrics: s.storageMetrics,
|
||||
Features: s.features,
|
||||
QOSQueue: s.queue,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.handler, err = grpcserver.ProvideService(s.cfg, s.features, interceptors.AuthenticatorFunc(s.authenticator), s.tracing, prometheus.DefaultRegisterer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
healthService, err := resource.ProvideHealthService(server)
|
||||
healthService, err := resource.ProvideHealthService(storageServer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
srv := s.handler.GetServer()
|
||||
resourcepb.RegisterResourceStoreServer(srv, server)
|
||||
resourcepb.RegisterBulkStoreServer(srv, server)
|
||||
resourcepb.RegisterResourceIndexServer(srv, server)
|
||||
resourcepb.RegisterManagedObjectIndexServer(srv, server)
|
||||
resourcepb.RegisterBlobStoreServer(srv, server)
|
||||
resourcepb.RegisterDiagnosticsServer(srv, server)
|
||||
resourcepb.RegisterQuotasServer(srv, server)
|
||||
// Register storage services
|
||||
resourcepb.RegisterResourceStoreServer(srv, storageServer)
|
||||
resourcepb.RegisterBulkStoreServer(srv, storageServer)
|
||||
resourcepb.RegisterBlobStoreServer(srv, storageServer)
|
||||
resourcepb.RegisterDiagnosticsServer(srv, storageServer)
|
||||
resourcepb.RegisterQuotasServer(srv, storageServer)
|
||||
// Register search services
|
||||
resourcepb.RegisterResourceIndexServer(srv, searchServer)
|
||||
resourcepb.RegisterManagedObjectIndexServer(srv, searchServer)
|
||||
grpc_health_v1.RegisterHealthServer(srv, healthService)
|
||||
|
||||
// register reflection service
|
||||
|
||||
@@ -0,0 +1,234 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"google.golang.org/grpc/health/grpc_health_v1"
|
||||
|
||||
"github.com/grafana/dskit/services"
|
||||
|
||||
infraDB "github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/modules"
|
||||
"github.com/grafana/grafana/pkg/services/authz"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/grpcserver"
|
||||
"github.com/grafana/grafana/pkg/services/grpcserver/interceptors"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resource/grpc"
|
||||
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
|
||||
"github.com/grafana/grafana/pkg/util/scheduler"
|
||||
)
|
||||
|
||||
var _ UnifiedStorageGrpcService = (*storageService)(nil)
|
||||
|
||||
type storageService struct {
|
||||
*services.BasicService
|
||||
|
||||
backend resource.StorageBackend
|
||||
cfg *setting.Cfg
|
||||
features featuremgmt.FeatureToggles
|
||||
db infraDB.DB
|
||||
stopCh chan struct{}
|
||||
stoppedCh chan error
|
||||
|
||||
handler grpcserver.Provider
|
||||
|
||||
tracing trace.Tracer
|
||||
|
||||
authenticator func(ctx context.Context) (context.Context, error)
|
||||
|
||||
log log.Logger
|
||||
reg prometheus.Registerer
|
||||
storageMetrics *resource.StorageMetrics
|
||||
|
||||
queue QOSEnqueueDequeuer
|
||||
scheduler *scheduler.Scheduler
|
||||
|
||||
// Subservices manager
|
||||
subservices *services.Manager
|
||||
subservicesWatcher *services.FailureWatcher
|
||||
hasSubservices bool
|
||||
}
|
||||
|
||||
func ProvideStorageService(
|
||||
cfg *setting.Cfg,
|
||||
features featuremgmt.FeatureToggles,
|
||||
db infraDB.DB,
|
||||
log log.Logger,
|
||||
reg prometheus.Registerer,
|
||||
storageMetrics *resource.StorageMetrics,
|
||||
backend resource.StorageBackend,
|
||||
) (UnifiedStorageGrpcService, error) {
|
||||
var err error
|
||||
tracer := otel.Tracer("unified-storage-server")
|
||||
|
||||
authn := NewAuthenticatorWithFallback(cfg, reg, tracer, func(ctx context.Context) (context.Context, error) {
|
||||
auth := grpc.Authenticator{Tracer: tracer}
|
||||
return auth.Authenticate(ctx)
|
||||
})
|
||||
|
||||
s := &storageService{
|
||||
backend: backend,
|
||||
cfg: cfg,
|
||||
features: features,
|
||||
stopCh: make(chan struct{}),
|
||||
stoppedCh: make(chan error, 1),
|
||||
authenticator: authn,
|
||||
tracing: tracer,
|
||||
db: db,
|
||||
log: log,
|
||||
reg: reg,
|
||||
storageMetrics: storageMetrics,
|
||||
subservicesWatcher: services.NewFailureWatcher(),
|
||||
}
|
||||
|
||||
subservices := []services.Service{}
|
||||
|
||||
// Setup QOS if enabled
|
||||
if cfg.QOSEnabled {
|
||||
qosReg := prometheus.WrapRegistererWithPrefix("resource_server_qos_", reg)
|
||||
queue := scheduler.NewQueue(&scheduler.QueueOptions{
|
||||
MaxSizePerTenant: cfg.QOSMaxSizePerTenant,
|
||||
Registerer: qosReg,
|
||||
})
|
||||
sched, err := scheduler.NewScheduler(queue, &scheduler.Config{
|
||||
NumWorkers: cfg.QOSNumberWorker,
|
||||
Logger: log,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create qos scheduler: %s", err)
|
||||
}
|
||||
|
||||
s.queue = queue
|
||||
s.scheduler = sched
|
||||
subservices = append(subservices, s.queue, s.scheduler)
|
||||
}
|
||||
|
||||
if len(subservices) > 0 {
|
||||
s.hasSubservices = true
|
||||
s.subservices, err = services.NewManager(subservices...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create subservices manager: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// This will be used when running as a dskit service
|
||||
s.BasicService = services.NewBasicService(s.starting, s.running, s.stopping).WithName(modules.StorageServer)
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *storageService) starting(ctx context.Context) error {
|
||||
if s.hasSubservices {
|
||||
s.subservicesWatcher.WatchManager(s.subservices)
|
||||
if err := services.StartManagerAndAwaitHealthy(ctx, s.subservices); err != nil {
|
||||
return fmt.Errorf("failed to start subservices: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
authzClient, err := authz.ProvideStandaloneAuthZClient(s.cfg, s.features, s.tracing, s.reg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Setup overrides service if enabled
|
||||
var overridesSvc *resource.OverridesService
|
||||
if s.cfg.OverridesFilePath != "" {
|
||||
overridesSvc, err = resource.NewOverridesService(context.Background(), s.log, s.reg, s.tracing, resource.ReloadOptions{
|
||||
FilePath: s.cfg.OverridesFilePath,
|
||||
ReloadPeriod: s.cfg.OverridesReloadInterval,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Create the storage server
|
||||
storageServer, err := NewStorageServer(StorageServerOptions{
|
||||
Backend: s.backend,
|
||||
OverridesService: overridesSvc,
|
||||
DB: s.db,
|
||||
Cfg: s.cfg,
|
||||
Tracer: s.tracing,
|
||||
Reg: s.reg,
|
||||
AccessClient: authzClient,
|
||||
StorageMetrics: s.storageMetrics,
|
||||
Features: s.features,
|
||||
QOSQueue: s.queue,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.handler, err = grpcserver.ProvideService(s.cfg, s.features, interceptors.AuthenticatorFunc(s.authenticator), s.tracing, prometheus.DefaultRegisterer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
healthService, err := resource.ProvideHealthService(storageServer)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
srv := s.handler.GetServer()
|
||||
// Register storage services
|
||||
resourcepb.RegisterResourceStoreServer(srv, storageServer)
|
||||
resourcepb.RegisterBulkStoreServer(srv, storageServer)
|
||||
resourcepb.RegisterBlobStoreServer(srv, storageServer)
|
||||
resourcepb.RegisterDiagnosticsServer(srv, storageServer)
|
||||
resourcepb.RegisterQuotasServer(srv, storageServer)
|
||||
grpc_health_v1.RegisterHealthServer(srv, healthService)
|
||||
|
||||
// register reflection service
|
||||
_, err = grpcserver.ProvideReflectionService(s.cfg, s.handler)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// start the gRPC server
|
||||
go func() {
|
||||
err := s.handler.Run(ctx)
|
||||
if err != nil {
|
||||
s.stoppedCh <- err
|
||||
} else {
|
||||
s.stoppedCh <- nil
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetAddress returns the address of the gRPC server.
|
||||
func (s *storageService) GetAddress() string {
|
||||
return s.handler.GetAddress()
|
||||
}
|
||||
|
||||
func (s *storageService) running(ctx context.Context) error {
|
||||
select {
|
||||
case err := <-s.stoppedCh:
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
return err
|
||||
}
|
||||
case err := <-s.subservicesWatcher.Chan():
|
||||
return fmt.Errorf("subservice failure: %w", err)
|
||||
case <-ctx.Done():
|
||||
close(s.stopCh)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *storageService) stopping(_ error) error {
|
||||
if s.hasSubservices {
|
||||
err := services.StopManagerAndAwaitStopped(context.Background(), s.subservices)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to stop subservices: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -32,6 +32,7 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource
|
||||
nsPrefix := "test-ns"
|
||||
|
||||
var server resource.ResourceServer
|
||||
var searchServer resource.SearchServer
|
||||
|
||||
t.Run("Create initial resources in storage", func(t *testing.T) {
|
||||
initialResources := []struct {
|
||||
@@ -96,25 +97,34 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource
|
||||
})
|
||||
|
||||
t.Run("Create a resource server with both backends", func(t *testing.T) {
|
||||
// Create a resource server with both backends
|
||||
// Create search server first
|
||||
var err error
|
||||
server, err = resource.NewResourceServer(resource.ResourceServerOptions{
|
||||
Backend: backend,
|
||||
Search: resource.SearchOptions{
|
||||
Backend: searchBackend,
|
||||
Resources: &resource.TestDocumentBuilderSupplier{
|
||||
GroupsResources: map[string]string{
|
||||
"test.grafana.app": "testresources",
|
||||
},
|
||||
searchOpts := resource.SearchOptions{
|
||||
Backend: searchBackend,
|
||||
Resources: &resource.TestDocumentBuilderSupplier{
|
||||
GroupsResources: map[string]string{
|
||||
"test.grafana.app": "testresources",
|
||||
},
|
||||
},
|
||||
}
|
||||
searchServer, err = resource.NewSearchServer(searchOpts, backend, nil, nil, nil, nil)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, searchServer)
|
||||
|
||||
// Initialize the search server
|
||||
err = searchServer.Init(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create a resource server separately from the search server
|
||||
server, err = resource.NewResourceServer(resource.ResourceServerOptions{
|
||||
Backend: backend,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
t.Run("Search for initial resources", func(t *testing.T) {
|
||||
// Test 1: Search for initial resources
|
||||
searchResp, err := server.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
searchResp, err := searchServer.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
Options: &resourcepb.ListOptions{
|
||||
Key: &resourcepb.ResourceKey{
|
||||
Group: "test.grafana.app",
|
||||
@@ -194,7 +204,7 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource
|
||||
})
|
||||
|
||||
t.Run("Search for documents", func(t *testing.T) {
|
||||
searchResp, err := server.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
searchResp, err := searchServer.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
Options: &resourcepb.ListOptions{
|
||||
Key: &resourcepb.ResourceKey{
|
||||
Group: "test.grafana.app",
|
||||
@@ -212,7 +222,7 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource
|
||||
})
|
||||
|
||||
t.Run("Search with tags", func(t *testing.T) {
|
||||
searchResp, err := server.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
searchResp, err := searchServer.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
Options: &resourcepb.ListOptions{
|
||||
Key: &resourcepb.ResourceKey{
|
||||
Group: "test.grafana.app",
|
||||
@@ -231,7 +241,7 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource
|
||||
require.Equal(t, int64(0), searchResp.TotalHits)
|
||||
|
||||
// this is the correct way of searching by tag
|
||||
searchResp, err = server.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
searchResp, err = searchServer.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
Options: &resourcepb.ListOptions{
|
||||
Key: &resourcepb.ResourceKey{
|
||||
Group: "test.grafana.app",
|
||||
@@ -253,7 +263,7 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource
|
||||
})
|
||||
|
||||
t.Run("Search with specific tag", func(t *testing.T) {
|
||||
searchResp, err := server.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
searchResp, err := searchServer.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
Options: &resourcepb.ListOptions{
|
||||
Key: &resourcepb.ResourceKey{
|
||||
Group: "test.grafana.app",
|
||||
@@ -272,7 +282,7 @@ func RunTestSearchAndStorage(t *testing.T, ctx context.Context, backend resource
|
||||
require.Equal(t, int64(0), searchResp.TotalHits)
|
||||
|
||||
// this is the correct way of searching by tag
|
||||
searchResp, err = server.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
searchResp, err = searchServer.Search(ctx, &resourcepb.ResourceSearchRequest{
|
||||
Options: &resourcepb.ListOptions{
|
||||
Key: &resourcepb.ResourceKey{
|
||||
Group: "test.grafana.app",
|
||||
|
||||
@@ -4,10 +4,15 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"math"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/dynamic"
|
||||
@@ -16,12 +21,167 @@ import (
|
||||
dashboardV0 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v0alpha1"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/identity"
|
||||
"github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
"github.com/grafana/grafana/pkg/services/dashboards"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/tests/apis"
|
||||
"github.com/grafana/grafana/pkg/tests/testinfra"
|
||||
"github.com/grafana/grafana/pkg/util/testutil"
|
||||
)
|
||||
|
||||
func TestIntegrationSearchDevDashboards(t *testing.T) {
|
||||
testutil.SkipIntegrationTestInShortMode(t)
|
||||
ctx := context.Background()
|
||||
|
||||
helper := apis.NewK8sTestHelper(t, testinfra.GrafanaOpts{
|
||||
DisableDataMigrations: true,
|
||||
AppModeProduction: true,
|
||||
DisableAnonymous: true,
|
||||
APIServerStorageType: "unified",
|
||||
UnifiedStorageConfig: map[string]setting.UnifiedStorageConfig{
|
||||
"dashboards.dashboard.grafana.app": {DualWriterMode: rest.Mode5},
|
||||
"folders.folder.grafana.app": {DualWriterMode: rest.Mode5},
|
||||
},
|
||||
UnifiedStorageEnableSearch: true,
|
||||
})
|
||||
defer helper.Shutdown()
|
||||
|
||||
// Create devenv dashboards from legacy API
|
||||
cfg := dynamic.ConfigFor(helper.Org1.Admin.NewRestConfig())
|
||||
cfg.GroupVersion = &dashboardV0.GroupVersion
|
||||
adminClient, err := k8srest.RESTClientFor(cfg)
|
||||
require.NoError(t, err)
|
||||
adminClient.Get()
|
||||
|
||||
fileCount := 0
|
||||
devenv := "../../../../devenv/dev-dashboards/panel-timeseries"
|
||||
err = filepath.WalkDir(devenv, func(p string, d fs.DirEntry, e error) error {
|
||||
require.NoError(t, err)
|
||||
if d.IsDir() || filepath.Ext(d.Name()) != ".json" {
|
||||
return nil
|
||||
}
|
||||
|
||||
// use the filename as UID
|
||||
uid := strings.TrimSuffix(d.Name(), ".json")
|
||||
if len(uid) > 40 {
|
||||
uid = uid[:40] // avoid uid too long, max 40 characters
|
||||
}
|
||||
|
||||
// nolint:gosec
|
||||
data, err := os.ReadFile(p)
|
||||
require.NoError(t, err)
|
||||
|
||||
cmd := dashboards.SaveDashboardCommand{
|
||||
Dashboard: &simplejson.Json{},
|
||||
Overwrite: true,
|
||||
}
|
||||
err = cmd.Dashboard.FromDB(data)
|
||||
require.NoError(t, err)
|
||||
cmd.Dashboard.Set("id", nil)
|
||||
cmd.Dashboard.Set("uid", uid)
|
||||
data, err = json.Marshal(cmd)
|
||||
require.NoError(t, err)
|
||||
|
||||
var statusCode int
|
||||
result := adminClient.Post().AbsPath("api", "dashboards", "db").
|
||||
Body(data).
|
||||
SetHeader("Content-type", "application/json").
|
||||
Do(ctx).
|
||||
StatusCode(&statusCode)
|
||||
require.NoError(t, result.Error(), "file: [%d] %s [status:%d]", fileCount, d.Name(), statusCode)
|
||||
require.Equal(t, int(http.StatusOK), statusCode)
|
||||
fileCount++
|
||||
return nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 16, fileCount, "file count from %s", devenv)
|
||||
|
||||
// Helper to call search
|
||||
callSearch := func(user apis.User, params string) dashboardV0.SearchResults {
|
||||
require.NotNil(t, user)
|
||||
ns := user.Identity.GetNamespace()
|
||||
cfg := dynamic.ConfigFor(user.NewRestConfig())
|
||||
cfg.GroupVersion = &dashboardV0.GroupVersion
|
||||
restClient, err := k8srest.RESTClientFor(cfg)
|
||||
require.NoError(t, err)
|
||||
|
||||
var statusCode int
|
||||
req := restClient.Get().AbsPath("apis", "dashboard.grafana.app", "v0alpha1", "namespaces", ns, "search").
|
||||
Param("limit", "1000").
|
||||
Param("type", "dashboard") // Only search dashboards
|
||||
|
||||
for kv := range strings.SplitSeq(params, "&") {
|
||||
if kv == "" {
|
||||
continue
|
||||
}
|
||||
parts := strings.SplitN(kv, "=", 2)
|
||||
if len(parts) == 2 {
|
||||
req = req.Param(parts[0], parts[1])
|
||||
}
|
||||
}
|
||||
res := req.Do(ctx).StatusCode(&statusCode)
|
||||
require.NoError(t, res.Error())
|
||||
require.Equal(t, int(http.StatusOK), statusCode)
|
||||
var sr dashboardV0.SearchResults
|
||||
raw, err := res.Raw()
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, json.Unmarshal(raw, &sr))
|
||||
|
||||
// Normalize scores and query cost for snapshot comparison
|
||||
sr.QueryCost = 0 // this depends on the hardware
|
||||
sr.MaxScore = roundTo(sr.MaxScore, 3)
|
||||
for i := range sr.Hits {
|
||||
sr.Hits[i].Score = roundTo(sr.Hits[i].Score, 3) // 0.6250571494814442 -> 0.625
|
||||
}
|
||||
return sr
|
||||
}
|
||||
|
||||
// Compare a results to snapshots
|
||||
testCases := []struct {
|
||||
name string
|
||||
user apis.User
|
||||
params string
|
||||
}{
|
||||
{
|
||||
name: "all",
|
||||
user: helper.Org1.Admin,
|
||||
params: "", // only dashboards
|
||||
},
|
||||
{
|
||||
name: "simple-query",
|
||||
user: helper.Org1.Admin,
|
||||
params: "query=stacking",
|
||||
},
|
||||
{
|
||||
name: "with-text-panel",
|
||||
user: helper.Org1.Admin,
|
||||
params: "field=panel_types&panelType=text",
|
||||
},
|
||||
}
|
||||
for i, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
res := callSearch(tc.user, tc.params)
|
||||
jj, err := json.MarshalIndent(res, "", " ")
|
||||
require.NoError(t, err)
|
||||
|
||||
fname := fmt.Sprintf("testdata/searchV0/t%02d-%s.json", i, tc.name)
|
||||
// nolint:gosec
|
||||
snapshot, err := os.ReadFile(fname)
|
||||
if err != nil {
|
||||
assert.Failf(t, "Failed to read snapshot", "file: %s", fname)
|
||||
err = os.WriteFile(fname, jj, 0o644)
|
||||
require.NoErrorf(t, err, "Failed to write snapshot file %s", fname)
|
||||
return
|
||||
}
|
||||
|
||||
if !assert.JSONEq(t, string(snapshot), string(jj)) {
|
||||
err = os.WriteFile(fname, jj, 0o644)
|
||||
require.NoErrorf(t, err, "Failed to write snapshot file %s", fname)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestIntegrationSearchPermissionFiltering(t *testing.T) {
|
||||
testutil.SkipIntegrationTestInShortMode(t)
|
||||
|
||||
@@ -285,3 +445,11 @@ func setFolderPermissions(t *testing.T, helper *apis.K8sTestHelper, actingUser a
|
||||
|
||||
require.Equal(t, http.StatusOK, resp.Response.StatusCode, "Failed to set permissions for folder %s", folderUID)
|
||||
}
|
||||
|
||||
// roundTo rounds a float64 to a specified number of decimal places.
|
||||
func roundTo(n float64, decimals uint32) float64 {
|
||||
// Calculate the power of 10 for the desired number of decimals
|
||||
scale := math.Pow(10, float64(decimals))
|
||||
// Multiply, round to the nearest integer, and then divide back
|
||||
return math.Round(n*scale) / scale
|
||||
}
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
{
|
||||
"totalHits": 16,
|
||||
"hits": [
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries",
|
||||
"title": "Panel Tests - Graph NG",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-by-value-color-schemes",
|
||||
"title": "Panel Tests - Graph NG - By value color schemes",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-nulls",
|
||||
"title": "Panel Tests - Graph NG - Discrete panels",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng",
|
||||
"timeseries",
|
||||
"trend",
|
||||
"state-timeline",
|
||||
"transform"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-gradient-area",
|
||||
"title": "Panel Tests - Graph NG - Gradient Area Fills",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-soft-limits",
|
||||
"title": "Panel Tests - Graph NG - softMin/softMax",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-yaxis-ticks",
|
||||
"title": "Panel Tests - Graph NG - Y axis ticks",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-hue-gradients",
|
||||
"title": "Panel Tests - GraphNG - Hue Gradients",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-time",
|
||||
"title": "Panel Tests - GraphNG - Time Axis",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-thresholds",
|
||||
"title": "Panel Tests - GraphNG Thresholds",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-shared-tooltip-cursor-positio",
|
||||
"title": "Panel Tests - shared tooltips cursor positioning",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-bars-high-density",
|
||||
"title": "Panel Tests - TimeSeries - bars high density (stroke + fill)",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-out-of-rage",
|
||||
"title": "Panel Tests - Timeseries - Out of range",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-stacking",
|
||||
"title": "Panel Tests - TimeSeries - stacking",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-formats",
|
||||
"title": "Panel Tests - Timeseries - Supported input formats"
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-stacking2",
|
||||
"title": "TimeSeries \u0026 BarChart Stacking",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-y-ticks-zero-decimals",
|
||||
"title": "Zero Decimals Y Ticks",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
]
|
||||
}
|
||||
],
|
||||
"maxScore": 1
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
{
|
||||
"totalHits": 2,
|
||||
"hits": [
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-stacking",
|
||||
"title": "Panel Tests - TimeSeries - stacking",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
],
|
||||
"score": 0.658
|
||||
},
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-stacking2",
|
||||
"title": "TimeSeries \u0026 BarChart Stacking",
|
||||
"tags": [
|
||||
"gdev",
|
||||
"panel-tests",
|
||||
"graph-ng"
|
||||
],
|
||||
"score": 0.625
|
||||
}
|
||||
],
|
||||
"maxScore": 0.658
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
{
|
||||
"totalHits": 1,
|
||||
"hits": [
|
||||
{
|
||||
"resource": "dashboards",
|
||||
"name": "timeseries-formats",
|
||||
"title": "Panel Tests - Timeseries - Supported input formats",
|
||||
"field": {
|
||||
"panel_types": [
|
||||
"table",
|
||||
"text",
|
||||
"timeseries"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"maxScore": 1.778
|
||||
}
|
||||
@@ -2054,9 +2054,7 @@ func TestIntegrationDeleteFolderWithProvisionedDashboards(t *testing.T) {
|
||||
DualWriterMode: modeDw,
|
||||
},
|
||||
},
|
||||
EnableFeatureToggles: []string{
|
||||
featuremgmt.FlagUnifiedStorageSearch,
|
||||
},
|
||||
UnifiedStorageEnableSearch: true,
|
||||
}
|
||||
|
||||
setupProvisioningDir(t, &ops)
|
||||
@@ -2163,9 +2161,7 @@ func TestIntegrationProvisionedFolderPropagatesLabelsAndAnnotations(t *testing.T
|
||||
DualWriterMode: mode3,
|
||||
},
|
||||
},
|
||||
EnableFeatureToggles: []string{
|
||||
featuremgmt.FlagUnifiedStorageSearch,
|
||||
},
|
||||
UnifiedStorageEnableSearch: true,
|
||||
}
|
||||
|
||||
setupProvisioningDir(t, &ops)
|
||||
|
||||
@@ -1830,6 +1830,22 @@
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "panelType",
|
||||
"in": "query",
|
||||
"description": "find dashboards using panels of a given plugin type",
|
||||
"schema": {
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "dataSourceType",
|
||||
"in": "query",
|
||||
"description": "find dashboards using datasources of a given plugin type",
|
||||
"schema": {
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "permission",
|
||||
"in": "query",
|
||||
|
||||
@@ -29,6 +29,7 @@
|
||||
"unicons/bookmark",
|
||||
"unicons/book-open",
|
||||
"unicons/brackets-curly",
|
||||
"unicons/brain",
|
||||
"unicons/bug",
|
||||
"unicons/building",
|
||||
"unicons/calculator-alt",
|
||||
|
||||
@@ -3,9 +3,19 @@ import { Unsubscribable } from 'rxjs';
|
||||
|
||||
import { getAppEvents } from '@grafana/runtime';
|
||||
import { useGrafana } from 'app/core/context/GrafanaContext';
|
||||
import { AbsoluteTimeEvent, CopyTimeEvent, PasteTimeEvent, ShiftTimeEvent, ZoomOutEvent } from 'app/types/events';
|
||||
import { getState } from 'app/store/store';
|
||||
import {
|
||||
AbsoluteTimeEvent,
|
||||
CopyTimeEvent,
|
||||
PasteTimeEvent,
|
||||
RunQueriesEvent,
|
||||
ShiftTimeEvent,
|
||||
ZoomOutEvent,
|
||||
} from 'app/types/events';
|
||||
import { useDispatch } from 'app/types/store';
|
||||
|
||||
import { runQueries } from '../state/query';
|
||||
import { selectPanesEntries } from '../state/selectors';
|
||||
import {
|
||||
copyTimeRangeToClipboard,
|
||||
makeAbsoluteTime,
|
||||
@@ -21,8 +31,23 @@ export function useKeyboardShortcuts() {
|
||||
useEffect(() => {
|
||||
keybindings.setupTimeRangeBindings(false);
|
||||
|
||||
// Explore-specific: run queries shortcut
|
||||
keybindings.bind('e r', () => {
|
||||
getAppEvents().publish(new RunQueriesEvent());
|
||||
});
|
||||
|
||||
const tearDown: Unsubscribable[] = [];
|
||||
|
||||
tearDown.push(
|
||||
getAppEvents().subscribe(RunQueriesEvent, () => {
|
||||
// Read panes at event time to avoid re-subscribing when panes change
|
||||
const panes = selectPanesEntries(getState());
|
||||
panes.forEach(([exploreId]) => {
|
||||
dispatch(runQueries({ exploreId }));
|
||||
});
|
||||
})
|
||||
);
|
||||
|
||||
tearDown.push(
|
||||
getAppEvents().subscribe(AbsoluteTimeEvent, () => {
|
||||
dispatch(makeAbsoluteTime());
|
||||
@@ -54,6 +79,7 @@ export function useKeyboardShortcuts() {
|
||||
);
|
||||
|
||||
return () => {
|
||||
keybindings.unbind('e r');
|
||||
tearDown.forEach((u) => u.unsubscribe());
|
||||
};
|
||||
}, [dispatch, keybindings]);
|
||||
|
||||
@@ -190,6 +190,22 @@ describe('preProcessLogs', () => {
|
||||
expect(logListModel.body).not.toBe(entry);
|
||||
});
|
||||
|
||||
test('Prettifies JSON with duplicate keys', () => {
|
||||
const entry = '{"key": "value", "key": "otherValue"}';
|
||||
const logListModel = createLogLine(
|
||||
{ entry },
|
||||
{
|
||||
escape: false,
|
||||
order: LogsSortOrder.Descending,
|
||||
timeZone: 'browser',
|
||||
wrapLogMessage: true, // wrapped
|
||||
prettifyJSON: true,
|
||||
}
|
||||
);
|
||||
expect(logListModel.entry).toBe(entry);
|
||||
expect(logListModel.body).not.toBe(entry);
|
||||
});
|
||||
|
||||
test('Prettifies and escapes wrapped JSON', () => {
|
||||
const entry = '{"key": "value", "otherKey": "other\\nValue"}';
|
||||
const logListModel = createLogLine(
|
||||
|
||||
@@ -135,7 +135,9 @@ export class LogListModel implements LogRowModel {
|
||||
get body(): string {
|
||||
if (this._body === undefined) {
|
||||
try {
|
||||
const parsed = parse(this.raw);
|
||||
const parsed = parse(this.raw, undefined, {
|
||||
onDuplicateKey: ({ newValue }) => newValue,
|
||||
});
|
||||
if (typeof parsed === 'object' && parsed !== null && !(parsed instanceof LosslessNumber)) {
|
||||
this._json = true;
|
||||
}
|
||||
|
||||
@@ -176,7 +176,7 @@ export function PluginDetailsBody({ plugin, queryParams, pageId, info, showDetai
|
||||
export const getStyles = (theme: GrafanaTheme2) => ({
|
||||
wrap: css({
|
||||
width: '100%',
|
||||
height: '50vh',
|
||||
height: '65vh',
|
||||
}),
|
||||
readme: css({
|
||||
'& img': {
|
||||
|
||||
@@ -62,6 +62,7 @@ export function PluginUsage({ plugin }: Props) {
|
||||
keyboardEvents={of()}
|
||||
onTagSelected={() => {}}
|
||||
trackingSource="PluginDetailsPage_PluginUsage"
|
||||
onClickItem={() => {}}
|
||||
/>
|
||||
);
|
||||
}}
|
||||
@@ -74,13 +75,13 @@ export function PluginUsage({ plugin }: Props) {
|
||||
return <Spinner />;
|
||||
}
|
||||
|
||||
if (!config.featureToggles.panelTitleSearch) {
|
||||
if (!config.featureToggles.unifiedStorageSearchUI) {
|
||||
return (
|
||||
<Alert
|
||||
title={t(
|
||||
'plugins.plugin-usage.title-missing-feature-toggle-panel-title-search',
|
||||
'Missing feature toggle: {{toggle}}',
|
||||
{ toggle: 'panelTitleSearch' }
|
||||
{ toggle: 'unifiedStorageSearchUI' }
|
||||
)}
|
||||
severity="warning"
|
||||
>
|
||||
|
||||
@@ -101,7 +101,7 @@ export const usePluginDetailsTabs = (
|
||||
}
|
||||
|
||||
if (
|
||||
config.featureToggles.panelTitleSearch &&
|
||||
config.featureToggles.unifiedStorageSearchUI &&
|
||||
(pluginConfig.meta.type === PluginType.panel || pluginConfig.meta.type === PluginType.datasource)
|
||||
) {
|
||||
navModelChildren.push({
|
||||
|
||||
@@ -25,7 +25,6 @@ export interface SearchQuery {
|
||||
sort?: string;
|
||||
ds_uid?: string;
|
||||
ds_type?: string;
|
||||
saved_query_uid?: string; // TODO: not implemented yet
|
||||
tags?: string[];
|
||||
kind?: string[];
|
||||
panel_type?: string;
|
||||
|
||||
@@ -297,6 +297,14 @@ export class UnifiedSearcher implements GrafanaSearcher {
|
||||
uri += '&' + query.kind.map((kind) => `type=${kind}`).join('&');
|
||||
}
|
||||
|
||||
if (query.ds_type?.length) {
|
||||
uri += '&dataSourceType=' + query.ds_type;
|
||||
}
|
||||
|
||||
if (query.panel_type?.length) {
|
||||
uri += '&panelType=' + query.panel_type;
|
||||
}
|
||||
|
||||
if (query.tags?.length) {
|
||||
uri += '&' + query.tags.map((tag) => `tag=${encodeURIComponent(tag)}`).join('&');
|
||||
}
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -555,7 +555,7 @@ export class TempoDatasource extends DataSourceWithBackend<TempoQuery, TempoJson
|
||||
|
||||
return {
|
||||
...expandedQuery,
|
||||
query: this.templateSrv.replace(query.query ?? '{}', scopedVars, VariableFormatID.Pipe),
|
||||
query: this.templateSrv.replace(query.query ?? '', scopedVars, VariableFormatID.Pipe),
|
||||
serviceMapQuery: Array.isArray(query.serviceMapQuery)
|
||||
? query.serviceMapQuery.map((query) => this.templateSrv.replace(query, scopedVars))
|
||||
: this.templateSrv.replace(query.serviceMapQuery ?? '', scopedVars),
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import config from '@grafana/plugin-configs/webpack.config.ts';
|
||||
|
||||
// eslint-disable-next-line no-barrel-files/no-barrel-files
|
||||
export default config;
|
||||
|
||||
@@ -159,6 +159,10 @@ export class AbsoluteTimeEvent extends BusEventWithPayload<AbsoluteTimeEventPayl
|
||||
static type = 'absolute-time';
|
||||
}
|
||||
|
||||
export class RunQueriesEvent extends BusEventBase {
|
||||
static type = 'run-queries';
|
||||
}
|
||||
|
||||
export class RemovePanelEvent extends BusEventWithPayload<number> {
|
||||
static type = 'remove-panel';
|
||||
}
|
||||
|
||||
@@ -19,6 +19,8 @@ export function getGrafanaContextMock(overrides: Partial<GrafanaContextType> = {
|
||||
clearAndInitGlobalBindings: jest.fn(),
|
||||
setupDashboardBindings: jest.fn(),
|
||||
setupTimeRangeBindings: jest.fn(),
|
||||
bind: jest.fn(),
|
||||
unbind: jest.fn(),
|
||||
} as unknown as KeybindingSrv,
|
||||
newAssetsChecker: {
|
||||
start: jest.fn(),
|
||||
|
||||
Reference in New Issue
Block a user