diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 2deae5a196e..f0ac47c3b60 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -107,6 +107,7 @@ /pkg/apimachinery/identity/ @grafana/identity-access-team /pkg/apimachinery/errutil/ @grafana/grafana-backend-group /pkg/promlib @grafana/observability-metrics +/pkg/storage/ @grafana/grafana-search-and-storage /pkg/services/annotations/ @grafana/grafana-search-and-storage /pkg/services/apikey/ @grafana/identity-access-team /pkg/services/cleanup/ @grafana/grafana-backend-group diff --git a/go.work.sum b/go.work.sum index 07147a44027..8843bb9859e 100644 --- a/go.work.sum +++ b/go.work.sum @@ -555,7 +555,6 @@ github.com/grafana/grafana/pkg/apimachinery v0.0.0-20240612101930-58f7032b3983/g github.com/grafana/grafana/pkg/promlib v0.0.3/go.mod h1:3El4NlsfALz8QQCbEGHGFvJUG+538QLMuALRhZ3pcoo= github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7 h1:pdN6V1QBWetyv/0+wjACpqVH+eVULgEjkurDLq3goeM= github.com/grpc-ecosystem/grpc-gateway/v2 v2.18.1/go.mod h1:YvJ2f6MplWDhfxiUC3KpyTy76kYUZA4W3pTv/wdKQ9Y= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 h1:bkypFPDjIYGfCYD5mRBvpqxfYX1YCS1PXdKYWi8FsN0= github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645 h1:MJG/KsmcqMwFAkh8mTnAwhyKoB+sTAnY4CACC110tbU= github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645/go.mod h1:6iZfnjpejD4L/4DwD7NryNaJyCQdzwWwH2MWhCA90Kw= github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed h1:5upAirOpQc1Q53c0bnx2ufif5kANL7bfZWcc6VJWJd8= @@ -961,7 +960,6 @@ google.golang.org/appengine v1.6.8 h1:IhEN5q69dyKagZPYMSdIjS2HqprW324FRQZJcGqPAs google.golang.org/genproto/googleapis/api v0.0.0-20240221002015-b0ce06bbee7c/go.mod h1:5iCWqnniDlqZHrd3neWVTOwvh/v6s3232omMecelax8= google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237/go.mod h1:Z5Iiy3jtmioajWHDGFk7CeugTyHtPvMHA4UTmUkyalE= google.golang.org/genproto/googleapis/api v0.0.0-20240513163218-0867130af1f8/go.mod h1:vPrPUTsDCYxXWjP7clS81mZ6/803D8K4iM9Ma27VKas= -google.golang.org/genproto/googleapis/api v0.0.0-20240604185151-ef581f913117 h1:+rdxYoE3E5htTEWIe15GlN6IfvbURM//Jt0mmkmm6ZU= google.golang.org/genproto/googleapis/bytestream v0.0.0-20231120223509-83a465c0220f h1:hL+1ptbhFoeL1HcROQ8OGXaqH0jYRRibgWQWco0/Ugc= google.golang.org/genproto/googleapis/bytestream v0.0.0-20231212172506-995d672761c0 h1:Y6QQt9D/syZt/Qgnz5a1y2O3WunQeeVDfS9+Xr82iFA= google.golang.org/genproto/googleapis/bytestream v0.0.0-20240125205218-1f4bbc51befe h1:weYsP+dNijSQVoLAb5bpUos3ciBpNU/NEVlHFKrk8pg= @@ -969,8 +967,8 @@ google.golang.org/genproto/googleapis/bytestream v0.0.0-20240325203815-454cdb8f5 google.golang.org/genproto/googleapis/rpc v0.0.0-20240221002015-b0ce06bbee7c/go.mod h1:H4O17MA/PE9BsGx3w+a+W2VOLLD1Qf7oJneAoU6WktY= google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY= google.golang.org/genproto/googleapis/rpc v0.0.0-20240513163218-0867130af1f8/go.mod h1:I7Y+G38R2bu5j1aLzfFmQfTcU/WnFuqDwLZAbvKTKpM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240520151616-dc85e6b867a5/go.mod h1:EfXuqaE1J41VCDicxHzUDm+8rk+7ZdXzHV0IhO/I6s0= google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157/go.mod h1:EfXuqaE1J41VCDicxHzUDm+8rk+7ZdXzHV0IhO/I6s0= -google.golang.org/genproto/googleapis/rpc v0.0.0-20240604185151-ef581f913117 h1:1GBuWVLM/KMVUv1t1En5Gs+gFZCNd360GGb4sSxtrhU= google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0 h1:M1YKkFIboKNieVO5DLUEVzQfGwJD30Nv2jfUgzb5UcE= gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/cheggaaa/pb.v1 v1.0.25 h1:Ev7yu1/f6+d+b3pi5vPdRPc6nNtP1umSfcWiEfRqv6I= diff --git a/pkg/apimachinery/auth/identity.go b/pkg/apimachinery/auth/identity.go deleted file mode 100644 index 8832b06d188..00000000000 --- a/pkg/apimachinery/auth/identity.go +++ /dev/null @@ -1 +0,0 @@ -package auth diff --git a/pkg/apimachinery/identity/context_test.go b/pkg/apimachinery/identity/context_test.go index 6464a7360fb..87f31a17763 100644 --- a/pkg/apimachinery/identity/context_test.go +++ b/pkg/apimachinery/identity/context_test.go @@ -17,126 +17,10 @@ func TestRequesterFromContext(t *testing.T) { }) t.Run("should return user set by ContextWithUser", func(t *testing.T) { - expected := &dummyUser{UID: "AAA"} + expected := &identity.StaticRequester{UserUID: "AAA"} ctx := identity.WithRequester(context.Background(), expected) actual, err := identity.GetRequester(ctx) require.NoError(t, err) require.Equal(t, expected.GetUID(), actual.GetUID()) }) } - -type dummyUser struct { - UID string -} - -// GetAuthID implements identity.Requester. -func (d *dummyUser) GetAuthID() string { - panic("unimplemented") -} - -// GetAuthenticatedBy implements identity.Requester. -func (d *dummyUser) GetAuthenticatedBy() string { - panic("unimplemented") -} - -// GetCacheKey implements identity.Requester. -func (d *dummyUser) GetCacheKey() string { - panic("unimplemented") -} - -// GetDisplayName implements identity.Requester. -func (d *dummyUser) GetDisplayName() string { - panic("unimplemented") -} - -// GetEmail implements identity.Requester. -func (d *dummyUser) GetEmail() string { - panic("unimplemented") -} - -// GetGlobalPermissions implements identity.Requester. -func (d *dummyUser) GetGlobalPermissions() map[string][]string { - panic("unimplemented") -} - -// GetID implements identity.Requester. -func (d *dummyUser) GetID() identity.NamespaceID { - panic("unimplemented") -} - -// GetIDToken implements identity.Requester. -func (d *dummyUser) GetIDToken() string { - panic("unimplemented") -} - -// GetIsGrafanaAdmin implements identity.Requester. -func (d *dummyUser) GetIsGrafanaAdmin() bool { - panic("unimplemented") -} - -// GetLogin implements identity.Requester. -func (d *dummyUser) GetLogin() string { - panic("unimplemented") -} - -// GetNamespacedID implements identity.Requester. -func (d *dummyUser) GetNamespacedID() (namespace identity.Namespace, identifier string) { - panic("unimplemented") -} - -// GetOrgID implements identity.Requester. -func (d *dummyUser) GetOrgID() int64 { - panic("unimplemented") -} - -// GetOrgName implements identity.Requester. -func (d *dummyUser) GetOrgName() string { - panic("unimplemented") -} - -// GetOrgRole implements identity.Requester. -func (d *dummyUser) GetOrgRole() identity.RoleType { - panic("unimplemented") -} - -// GetPermissions implements identity.Requester. -func (d *dummyUser) GetPermissions() map[string][]string { - panic("unimplemented") -} - -// GetTeams implements identity.Requester. -func (d *dummyUser) GetTeams() []int64 { - panic("unimplemented") -} - -// GetUID implements identity.Requester. -func (d *dummyUser) GetUID() identity.NamespaceID { - return identity.NewNamespaceIDString(identity.NamespaceUser, d.UID) -} - -// HasRole implements identity.Requester. -func (d *dummyUser) HasRole(role identity.RoleType) bool { - panic("unimplemented") -} - -// HasUniqueId implements identity.Requester. -func (d *dummyUser) HasUniqueId() bool { - panic("unimplemented") -} - -// IsAuthenticatedBy implements identity.Requester. -func (d *dummyUser) IsAuthenticatedBy(providers ...string) bool { - panic("unimplemented") -} - -// IsEmailVerified implements identity.Requester. -func (d *dummyUser) IsEmailVerified() bool { - panic("unimplemented") -} - -// IsNil implements identity.Requester. -func (d *dummyUser) IsNil() bool { - return false -} - -var _ identity.Requester = &dummyUser{} diff --git a/pkg/apimachinery/identity/static.go b/pkg/apimachinery/identity/static.go new file mode 100644 index 00000000000..b7691b2e286 --- /dev/null +++ b/pkg/apimachinery/identity/static.go @@ -0,0 +1,176 @@ +package identity + +import "fmt" + +var _ Requester = &StaticRequester{} + +// StaticRequester is helpful for tests +// This is mostly copied from: +// https://github.com/grafana/grafana/blob/v11.0.0/pkg/services/user/identity.go#L16 +type StaticRequester struct { + Namespace Namespace + UserID int64 + UserUID string + OrgID int64 + OrgName string + OrgRole RoleType + Login string + Name string + DisplayName string + Email string + EmailVerified bool + AuthID string + AuthenticatedBy string + IsGrafanaAdmin bool + IsAnonymous bool + IsDisabled bool + // Permissions grouped by orgID and actions + Permissions map[int64]map[string][]string + IDToken string + CacheKey string +} + +func (u *StaticRequester) HasRole(role RoleType) bool { + if u.IsGrafanaAdmin { + return true + } + + return u.OrgRole.Includes(role) +} + +// GetIsGrafanaAdmin returns true if the user is a server admin +func (u *StaticRequester) GetIsGrafanaAdmin() bool { + return u.IsGrafanaAdmin +} + +// GetLogin returns the login of the active entity +// Can be empty if the user is anonymous +func (u *StaticRequester) GetLogin() string { + return u.Login +} + +// GetOrgID returns the ID of the active organization +func (u *StaticRequester) GetOrgID() int64 { + return u.OrgID +} + +// DEPRECATED: GetOrgName returns the name of the active organization +// Retrieve the organization name from the organization service instead of using this method. +func (u *StaticRequester) GetOrgName() string { + return u.OrgName +} + +// GetPermissions returns the permissions of the active entity +func (u *StaticRequester) GetPermissions() map[string][]string { + if u.Permissions == nil { + return make(map[string][]string) + } + + if u.Permissions[u.GetOrgID()] == nil { + return make(map[string][]string) + } + + return u.Permissions[u.GetOrgID()] +} + +// GetGlobalPermissions returns the permissions of the active entity that are available across all organizations +func (u *StaticRequester) GetGlobalPermissions() map[string][]string { + if u.Permissions == nil { + return make(map[string][]string) + } + + const globalOrgID = 0 + + if u.Permissions[globalOrgID] == nil { + return make(map[string][]string) + } + + return u.Permissions[globalOrgID] +} + +// DEPRECATED: GetTeams returns the teams the entity is a member of +// Retrieve the teams from the team service instead of using this method. +func (u *StaticRequester) GetTeams() []int64 { + return []int64{} // Not implemented +} + +// GetOrgRole returns the role of the active entity in the active organization +func (u *StaticRequester) GetOrgRole() RoleType { + return u.OrgRole +} + +// HasUniqueId returns true if the entity has a unique id +func (u *StaticRequester) HasUniqueId() bool { + return u.UserID > 0 +} + +// GetID returns namespaced id for the entity +func (u *StaticRequester) GetID() NamespaceID { + return NewNamespaceIDString(u.Namespace, fmt.Sprintf("%d", u.UserID)) +} + +// GetUID returns namespaced uid for the entity +func (u *StaticRequester) GetUID() NamespaceID { + return NewNamespaceIDString(u.Namespace, u.UserUID) +} + +// GetNamespacedID returns the namespace and ID of the active entity +// The namespace is one of the constants defined in pkg/apimachinery/identity +func (u *StaticRequester) GetNamespacedID() (Namespace, string) { + return u.Namespace, fmt.Sprintf("%d", u.UserID) +} + +func (u *StaticRequester) GetAuthID() string { + return u.AuthID +} + +func (u *StaticRequester) GetAuthenticatedBy() string { + return u.AuthenticatedBy +} + +func (u *StaticRequester) IsAuthenticatedBy(providers ...string) bool { + for _, p := range providers { + if u.AuthenticatedBy == p { + return true + } + } + return false +} + +// FIXME: remove this method once all services are using an interface +func (u *StaticRequester) IsNil() bool { + return u == nil +} + +// GetEmail returns the email of the active entity +// Can be empty. +func (u *StaticRequester) GetEmail() string { + return u.Email +} + +func (u *StaticRequester) IsEmailVerified() bool { + return u.EmailVerified +} + +func (u *StaticRequester) GetCacheKey() string { + return u.CacheKey +} + +// GetDisplayName returns the display name of the active entity +// The display name is the name if it is set, otherwise the login or email +func (u *StaticRequester) GetDisplayName() string { + if u.DisplayName != "" { + return u.DisplayName + } + if u.Name != "" { + return u.Name + } + if u.Login != "" { + return u.Login + } + return u.Email +} + +func (u *StaticRequester) GetIDToken() string { + return u.IDToken +} diff --git a/pkg/storage/unified/resource/resource.proto b/pkg/storage/unified/resource/resource.proto index 11a140ad5d5..810279e3ded 100644 --- a/pkg/storage/unified/resource/resource.proto +++ b/pkg/storage/unified/resource/resource.proto @@ -422,9 +422,11 @@ service ResourceStore { rpc IsHealthy(HealthCheckRequest) returns (HealthCheckResponse); } -// Unlike ResourceStore, this service does not have strict read after write guarantees // Clients can use this service directly +// NOTE: This is read only, and no read afer write guarantees service ResourceSearch { + // rpc Search(...) ... + rpc Read(ReadRequest) returns (ReadResponse); // Duplicated -- for client read only usage // Get the raw blob bytes and metadata diff --git a/pkg/storage/unified/resource/writer_test.go b/pkg/storage/unified/resource/writer_test.go index 64daaac71b9..55db3bfff6c 100644 --- a/pkg/storage/unified/resource/writer_test.go +++ b/pkg/storage/unified/resource/writer_test.go @@ -13,18 +13,18 @@ import ( "github.com/grafana/grafana/pkg/apimachinery/identity" "github.com/grafana/grafana/pkg/apimachinery/utils" - "github.com/grafana/grafana/pkg/infra/appcontext" - "github.com/grafana/grafana/pkg/services/user" ) func TestWriter(t *testing.T) { tracer := noop.NewTracerProvider().Tracer("testing") - testUserA := &user.SignedInUser{ - UserID: 123, - UserUID: "u123", - OrgRole: identity.RoleAdmin, + testUserA := &identity.StaticRequester{ + Namespace: identity.NamespaceUser, + UserID: 123, + UserUID: "u123", + OrgRole: identity.RoleAdmin, + IsGrafanaAdmin: true, // can do anything } - ctx := appcontext.WithUser(context.Background(), testUserA) + ctx := identity.WithRequester(context.Background(), testUserA) store := NewMemoryStore() writer, err := NewResourceWriter(WriterOptions{ diff --git a/pkg/storage/unified/sqlstash/queries.go b/pkg/storage/unified/sqlstash/queries.go deleted file mode 100644 index 2777702ce9c..00000000000 --- a/pkg/storage/unified/sqlstash/queries.go +++ /dev/null @@ -1,222 +0,0 @@ -package sqlstash - -import ( - "context" - "database/sql" - "embed" - "errors" - "fmt" - "strings" - "text/template" - - "github.com/grafana/grafana/pkg/services/store/entity/db" - "github.com/grafana/grafana/pkg/services/store/entity/sqlstash/sqltemplate" -) - -// Templates setup. -var ( - //go:embed data/*.sql - sqlTemplatesFS embed.FS - - // all templates - helpers = template.FuncMap{ - "listSep": helperListSep, - "join": helperJoin, - } - sqlTemplates = template.Must(template.New("sql").Funcs(helpers).ParseFS(sqlTemplatesFS, `data/*.sql`)) -) - -func mustTemplate(filename string) *template.Template { - if t := sqlTemplates.Lookup(filename); t != nil { - return t - } - panic(fmt.Sprintf("template file not found: %s", filename)) -} - -// Templates. -var ( - sqlResourceVersionGet = mustTemplate("rv_get.sql") - sqlResourceVersionInc = mustTemplate("rv_inc.sql") - sqlResourceVersionInsert = mustTemplate("rv_insert.sql") - sqlResourceVersionLock = mustTemplate("rv_lock.sql") - - sqlResourceInsert = mustTemplate("resource_insert.sql") - sqlResourceGet = mustTemplate("resource_get.sql") -) - -// TxOptions. -var ( - ReadCommitted = &sql.TxOptions{ - Isolation: sql.LevelReadCommitted, - } - ReadCommittedRO = &sql.TxOptions{ - Isolation: sql.LevelReadCommitted, - ReadOnly: true, - } -) - -// SQLError is an error returned by the database, which includes additionally -// debugging information about what was sent to the database. -type SQLError struct { - Err error - CallType string // either Query, QueryRow or Exec - TemplateName string - Query string - RawQuery string - ScanDest []any - - // potentially regulated information is not exported and only directly - // available for local testing and local debugging purposes, making sure it - // is never marshaled to JSON or any other serialization. - - arguments []any -} - -func (e SQLError) Unwrap() error { - return e.Err -} - -func (e SQLError) Error() string { - return fmt.Sprintf("%s: %s with %d input arguments and %d output "+ - "destination arguments: %v", e.TemplateName, e.CallType, - len(e.arguments), len(e.ScanDest), e.Err) -} - -//------------------------------------------------------------------------ -// Resource Version table support -//------------------------------------------------------------------------ - -type returnsResourceVersion struct { - ResourceVersion int64 -} - -func (r *returnsResourceVersion) Results() (*returnsResourceVersion, error) { - return r, nil -} - -type sqlResourceVersionGetRequest struct { - *sqltemplate.SQLTemplate - Group string - Resource string - *returnsResourceVersion -} - -func (r sqlResourceVersionGetRequest) Validate() error { - return nil // TODO -} - -type sqlResourceVersionLockRequest struct { - *sqltemplate.SQLTemplate - Group string - Resource string - *returnsResourceVersion -} - -func (r sqlResourceVersionLockRequest) Validate() error { - return nil // TODO -} - -type sqlResourceVersionIncRequest struct { - *sqltemplate.SQLTemplate - Group string - Resource string - ResourceVersion int64 -} - -func (r sqlResourceVersionIncRequest) Validate() error { - return nil // TODO -} - -type sqlResourceVersionInsertRequest struct { - *sqltemplate.SQLTemplate - Group string - Resource string -} - -func (r sqlResourceVersionInsertRequest) Validate() error { - return nil // TODO -} - -// resourceVersionAtomicInc atomically increases the version of a kind within a -// transaction. -func resourceVersionAtomicInc(ctx context.Context, x db.ContextExecer, d sqltemplate.Dialect, group, resource string) (newVersion int64, err error) { - // 1. Lock the kind and get the latest version - lockReq := sqlResourceVersionLockRequest{ - SQLTemplate: sqltemplate.New(d), - Group: group, - Resource: resource, - returnsResourceVersion: new(returnsResourceVersion), - } - kindv, err := queryRow(ctx, x, sqlResourceVersionLock, lockReq) - - // if there wasn't a row associated with the given kind, we create one with - // version 1 - if errors.Is(err, sql.ErrNoRows) { - // NOTE: there is a marginal chance that we race with another writer - // trying to create the same row. This is only possible when onboarding - // a new (Group, Resource) to the cell, which should be very unlikely, - // and the workaround is simply retrying. The alternative would be to - // use INSERT ... ON CONFLICT DO UPDATE ..., but that creates a - // requirement for support in Dialect only for this marginal case, and - // we would rather keep Dialect as small as possible. Another - // alternative is to simply check if the INSERT returns a DUPLICATE KEY - // error and then retry the original SELECT, but that also adds some - // complexity to the code. That would be preferrable to changing - // Dialect, though. The current alternative, just retrying, seems to be - // enough for now. - insReq := sqlResourceVersionInsertRequest{ - SQLTemplate: sqltemplate.New(d), - Group: group, - Resource: resource, - } - if _, err = exec(ctx, x, sqlResourceVersionInsert, insReq); err != nil { - return 0, fmt.Errorf("insert into kind_version: %w", err) - } - - return 1, nil - } - - if err != nil { - return 0, fmt.Errorf("lock kind: %w", err) - } - - incReq := sqlResourceVersionIncRequest{ - SQLTemplate: sqltemplate.New(d), - Group: group, - Resource: resource, - ResourceVersion: kindv.ResourceVersion, - } - if _, err = exec(ctx, x, sqlResourceVersionInc, incReq); err != nil { - return 0, fmt.Errorf("increase kind version: %w", err) - } - - return kindv.ResourceVersion + 1, nil -} - -// Template helpers. - -// helperListSep is a helper that helps writing simpler loops in SQL templates. -// Example usage: -// -// {{ $comma := listSep ", " }} -// {{ range .Values }} -// {{/* here we put "-" on each end to remove extra white space */}} -// {{- call $comma -}} -// {{ .Value }} -// {{ end }} -func helperListSep(sep string) func() string { - var addSep bool - - return func() string { - if addSep { - return sep - } - addSep = true - - return "" - } -} - -func helperJoin(sep string, elems ...string) string { - return strings.Join(elems, sep) -} diff --git a/pkg/storage/unified/sqlstash/sql_storage_server.go b/pkg/storage/unified/sqlstash/sql_storage_server.go index a9b187d3bbe..403f964775b 100644 --- a/pkg/storage/unified/sqlstash/sql_storage_server.go +++ b/pkg/storage/unified/sqlstash/sql_storage_server.go @@ -2,7 +2,6 @@ package sqlstash import ( "context" - "database/sql" "errors" "fmt" "strings" @@ -23,12 +22,7 @@ import ( // Package-level errors. var ( - ErrNotFound = errors.New("entity not found") - ErrOptimisticLockingFailed = errors.New("optimistic locking failed") - ErrUserNotFoundInContext = errors.New("user not found in context") - ErrNextPageTokenNotSupported = errors.New("nextPageToken not yet supported") - ErrLimitNotSupported = errors.New("limit not yet supported") - ErrNotImplementedYet = errors.New("not implemented yet") + ErrNotImplementedYet = errors.New("not implemented yet") ) // Make sure we implement both store and search @@ -38,6 +32,7 @@ var _ resource.ResourceSearchServer = &sqlResourceServer{} func ProvideSQLResourceServer(db db.EntityDBInterface, tracer tracing.Tracer) (SqlResourceServer, error) { ctx, cancel := context.WithCancel(context.Background()) + var err error server := &sqlResourceServer{ db: db, log: log.New("sql-resource-server"), @@ -45,6 +40,15 @@ func ProvideSQLResourceServer(db db.EntityDBInterface, tracer tracing.Tracer) (S cancel: cancel, tracer: tracer, } + server.writer, err = resource.NewResourceWriter(resource.WriterOptions{ + NodeID: 123, // for snowflake ID generation + Tracer: tracer, + Reader: server.Read, + Appender: server.append, + }) + if err != nil { + return nil, err + } if err := prometheus.Register(sqlstash.NewStorageMetrics()); err != nil { server.log.Warn("error registering storage server metrics", "error", err) @@ -55,6 +59,7 @@ func ProvideSQLResourceServer(db db.EntityDBInterface, tracer tracing.Tracer) (S type SqlResourceServer interface { resource.ResourceStoreServer + resource.ResourceSearchServer Init() error Stop() @@ -70,7 +75,9 @@ type sqlResourceServer struct { cancel context.CancelFunc stream chan *resource.WatchResponse tracer trace.Tracer - writer resource.ResourceWriter + + // Wrapper around all write events + writer resource.ResourceWriter once sync.Once initErr error @@ -184,6 +191,8 @@ func (s *sqlResourceServer) append(ctx context.Context, event *resource.WriteEve _, span := s.tracer.Start(ctx, "storage_server.WriteEvent") defer span.End() + // TODO... actually write write the event! + return 0, ErrNotImplementedYet } @@ -244,42 +253,21 @@ func (s *sqlResourceServer) Delete(ctx context.Context, req *resource.DeleteRequ } func (s *sqlResourceServer) List(ctx context.Context, req *resource.ListRequest) (*resource.ListResponse, error) { - ctx, span := s.tracer.Start(ctx, "storage_server.List") + _, span := s.tracer.Start(ctx, "storage_server.List") defer span.End() if err := s.Init(); err != nil { return nil, err } - var rv int64 - err := s.sqlDB.WithTx(ctx, ReadCommitted, func(ctx context.Context, tx db.Tx) error { - req := sqlResourceVersionGetRequest{ - SQLTemplate: sqltemplate.New(s.sqlDialect), - Group: req.Options.Key.Group, - Resource: req.Options.Key.Resource, - returnsResourceVersion: new(returnsResourceVersion), - } - res, err := queryRow(ctx, tx, sqlResourceVersionGet, req) - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return err - } - if res != nil { - rv = res.ResourceVersion - } - return nil - }) - if err != nil { - return nil, err - } - - fmt.Printf("TODO, LIST: %+v // %d", req.Options.Key, rv) + fmt.Printf("TODO, LIST: %+v", req.Options.Key) return nil, ErrNotImplementedYet } // Get the raw blob bytes and metadata func (s *sqlResourceServer) GetBlob(ctx context.Context, req *resource.GetBlobRequest) (*resource.GetBlobResponse, error) { - ctx, span := s.tracer.Start(ctx, "storage_server.List") + _, span := s.tracer.Start(ctx, "storage_server.List") defer span.End() if err := s.Init(); err != nil { @@ -293,7 +281,7 @@ func (s *sqlResourceServer) GetBlob(ctx context.Context, req *resource.GetBlobRe // Show resource history (and trash) func (s *sqlResourceServer) History(ctx context.Context, req *resource.HistoryRequest) (*resource.HistoryResponse, error) { - ctx, span := s.tracer.Start(ctx, "storage_server.History") + _, span := s.tracer.Start(ctx, "storage_server.History") defer span.End() if err := s.Init(); err != nil { @@ -307,7 +295,7 @@ func (s *sqlResourceServer) History(ctx context.Context, req *resource.HistoryRe // Used for efficient provisioning func (s *sqlResourceServer) Origin(ctx context.Context, req *resource.OriginRequest) (*resource.OriginResponse, error) { - ctx, span := s.tracer.Start(ctx, "storage_server.History") + _, span := s.tracer.Start(ctx, "storage_server.History") defer span.End() if err := s.Init(); err != nil { diff --git a/pkg/storage/unified/sqlstash/testdata/01_create_playlist.json b/pkg/storage/unified/sqlstash/testdata/01_create_playlist.json deleted file mode 100644 index b0272532e8d..00000000000 --- a/pkg/storage/unified/sqlstash/testdata/01_create_playlist.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "apiVersion": "playlist.grafana.app/v0alpha1", - "kind": "Playlist", - "metadata": { - "name": "fdgsv37qslr0ga", - "namespace": "default", - "annotations": { - "grafana.app/originName": "elsewhere", - "grafana.app/originPath": "path/to/item", - "grafana.app/originTimestamp": "2024-02-02T00:00:00Z" - }, - "creationTimestamp": "2024-03-03T00:00:00Z", - "uid": "8tGrXJgGbFI0" - }, - "spec": { - "title": "hello", - "interval": "5m", - "items": [ - { - "type": "dashboard_by_uid", - "value": "vmie2cmWz" - } - ] - } -} \ No newline at end of file diff --git a/pkg/storage/unified/sqlstash/utils.go b/pkg/storage/unified/sqlstash/utils.go index e50bb3742c8..afc742fd766 100644 --- a/pkg/storage/unified/sqlstash/utils.go +++ b/pkg/storage/unified/sqlstash/utils.go @@ -1,14 +1,8 @@ package sqlstash import ( - "context" - "database/sql" - "fmt" "net/http" - "text/template" - "github.com/grafana/grafana/pkg/services/store/entity/db" - "github.com/grafana/grafana/pkg/services/store/entity/sqlstash/sqltemplate" "github.com/grafana/grafana/pkg/storage/unified/resource" ) @@ -19,118 +13,3 @@ func badRequest(msg string) *resource.StatusResult { Code: http.StatusBadRequest, } } - -// ptrOr returns the first non-nil pointer in the list or a new non-nil pointer. -func ptrOr[P ~*E, E any](ps ...P) P { - for _, p := range ps { - if p != nil { - return p - } - } - - return P(new(E)) -} - -// sliceOr returns the first slice that has at least one element, or a new empty -// slice. -func sliceOr[S ~[]E, E comparable](vals ...S) S { - for _, s := range vals { - if len(s) > 0 { - return s - } - } - - return S{} -} - -// mapOr returns the first map that has at least one element, or a new empty -// map. -func mapOr[M ~map[K]V, K comparable, V any](vals ...M) M { - for _, m := range vals { - if len(m) > 0 { - return m - } - } - - return M{} -} - -// queryRow uses `req` as input and output for a single-row returning query -// generated with `tmpl`, and executed in `x`. -func queryRow[T any](ctx context.Context, x db.ContextExecer, tmpl *template.Template, req sqltemplate.WithResults[T]) (T, error) { - var zero T - - if err := req.Validate(); err != nil { - return zero, fmt.Errorf("query: invalid request for template %q: %w", - tmpl.Name(), err) - } - - rawQuery, err := sqltemplate.Execute(tmpl, req) - if err != nil { - return zero, fmt.Errorf("execute template: %w", err) - } - query := sqltemplate.FormatSQL(rawQuery) - - row := x.QueryRowContext(ctx, query, req.GetArgs()...) - if err := row.Err(); err != nil { - return zero, SQLError{ - Err: err, - CallType: "QueryRow", - TemplateName: tmpl.Name(), - arguments: req.GetArgs(), - ScanDest: req.GetScanDest(), - Query: query, - RawQuery: rawQuery, - } - } - - return scanRow(row, req) -} - -type scanner interface { - Scan(dest ...any) error -} - -// scanRow is used on *sql.Row and *sql.Rows, and is factored out here not to -// improving code reuse, but rather for ease of testing. -func scanRow[T any](sc scanner, req sqltemplate.WithResults[T]) (zero T, err error) { - if err = sc.Scan(req.GetScanDest()...); err != nil { - return zero, fmt.Errorf("row scan: %w", err) - } - - res, err := req.Results() - if err != nil { - return zero, fmt.Errorf("row results: %w", err) - } - - return res, nil -} - -// exec uses `req` as input for a non-data returning query generated with -// `tmpl`, and executed in `x`. -func exec(ctx context.Context, x db.ContextExecer, tmpl *template.Template, req sqltemplate.SQLTemplateIface) (sql.Result, error) { - if err := req.Validate(); err != nil { - return nil, fmt.Errorf("exec: invalid request for template %q: %w", - tmpl.Name(), err) - } - - rawQuery, err := sqltemplate.Execute(tmpl, req) - if err != nil { - return nil, fmt.Errorf("execute template: %w", err) - } - query := sqltemplate.FormatSQL(rawQuery) - - res, err := x.ExecContext(ctx, query, req.GetArgs()...) - if err != nil { - return nil, SQLError{ - Err: err, - CallType: "Exec", - TemplateName: tmpl.Name(), - arguments: req.GetArgs(), - Query: query, - RawQuery: rawQuery, - } - } - - return res, nil -} diff --git a/pkg/storage/unified/sqlstash/utils_test.go b/pkg/storage/unified/sqlstash/utils_test.go deleted file mode 100644 index 313be4219aa..00000000000 --- a/pkg/storage/unified/sqlstash/utils_test.go +++ /dev/null @@ -1,503 +0,0 @@ -package sqlstash - -import ( - "database/sql" - "database/sql/driver" - "errors" - "fmt" - "io" - "regexp" - "strings" - "testing" - "text/template" - - sqlmock "github.com/DATA-DOG/go-sqlmock" - "github.com/stretchr/testify/require" - - "github.com/grafana/grafana/pkg/services/store/entity/db" - "github.com/grafana/grafana/pkg/services/store/entity/db/dbimpl" - "github.com/grafana/grafana/pkg/services/store/entity/sqlstash/sqltemplate" - sqltemplateMocks "github.com/grafana/grafana/pkg/services/store/entity/sqlstash/sqltemplate/mocks" - "github.com/grafana/grafana/pkg/util/testutil" -) - -// newMockDBNopSQL returns a db.DB and a sqlmock.Sqlmock that doesn't validates -// SQL. This is only meant to be used to test wrapping utilities exec, query and -// queryRow, where the actual SQL is not relevant to the unit tests, but rather -// how the possible derived error conditions handled. -func newMockDBNopSQL(t *testing.T) (db.DB, sqlmock.Sqlmock) { - t.Helper() - - db, mock, err := sqlmock.New( - sqlmock.MonitorPingsOption(true), - sqlmock.QueryMatcherOption(sqlmock.QueryMatcherFunc( - func(expectedSQL, actualSQL string) error { - return nil - }, - )), - ) - - return newUnitTestDB(t, db, mock, err) -} - -// newMockDBMatchWords returns a db.DB and a sqlmock.Sqlmock that will match SQL -// by splitting the expected SQL string into words, and then try to find all of -// them in the actual SQL, in the given order, case insensitively. Prepend a -// word with a `!` to say that word should not be found. -func newMockDBMatchWords(t *testing.T) (db.DB, sqlmock.Sqlmock) { - t.Helper() - - db, mock, err := sqlmock.New( - sqlmock.MonitorPingsOption(true), - sqlmock.QueryMatcherOption( - sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error { - actualSQL = strings.ToLower(sqltemplate.FormatSQL(actualSQL)) - expectedSQL = strings.ToLower(expectedSQL) - - var offset int - for _, vv := range mockDBMatchWordsRE.FindAllStringSubmatch(expectedSQL, -1) { - v := vv[1] - - var shouldNotMatch bool - if v != "" && v[0] == '!' { - v = v[1:] - shouldNotMatch = true - } - if v == "" { - return fmt.Errorf("invalid expected word %q in %q", v, - expectedSQL) - } - - reWord, err := regexp.Compile(`\b` + regexp.QuoteMeta(v) + `\b`) - if err != nil { - return fmt.Errorf("compile word %q from expected SQL: %s", v, - expectedSQL) - } - - if shouldNotMatch { - if reWord.MatchString(actualSQL[offset:]) { - return fmt.Errorf("actual SQL fragent should not cont"+ - "ain %q but it does\n\tFragment: %s\n\tFull SQL: %s", - v, actualSQL[offset:], actualSQL) - } - } else { - loc := reWord.FindStringIndex(actualSQL[offset:]) - if len(loc) == 0 { - return fmt.Errorf("actual SQL fragment should contain "+ - "%q but it doesn't\n\tFragment: %s\n\tFull SQL: %s", - v, actualSQL[offset:], actualSQL) - } - offset = loc[1] // advance the offset - } - } - - return nil - }, - ), - ), - ) - - return newUnitTestDB(t, db, mock, err) -} - -var mockDBMatchWordsRE = regexp.MustCompile(`(?:\W|\A)(!?\w+)\b`) - -func newUnitTestDB(t *testing.T, db *sql.DB, mock sqlmock.Sqlmock, err error) (db.DB, sqlmock.Sqlmock) { - t.Helper() - - require.NoError(t, err) - - return dbimpl.NewDB(db, "sqlmock"), mock -} - -// mockResults aids in testing code paths with queries returning large number of -// values, like those returning *entity.Entity. This is because we want to -// emulate returning the same row columns and row values the same as a real -// database would do. This utility the same template SQL that is expected to be -// used to help populate all the expected fields. -// fileds -type mockResults[T any] struct { - t *testing.T - tmpl *template.Template - data sqltemplate.WithResults[T] - rows *sqlmock.Rows -} - -// newMockResults returns a new *mockResults. If you want to emulate a call -// returning zero rows, then immediately call the Row method afterward. -func newMockResults[T any](t *testing.T, mock sqlmock.Sqlmock, tmpl *template.Template, data sqltemplate.WithResults[T]) *mockResults[T] { - t.Helper() - - data.Reset() - err := tmpl.Execute(io.Discard, data) - require.NoError(t, err) - rows := mock.NewRows(data.GetColNames()) - - return &mockResults[T]{ - t: t, - tmpl: tmpl, - data: data, - rows: rows, - } -} - -// AddCurrentData uses the values contained in the `data` argument used during -// creation to populate a new expected row. It will access `data` with pointers, -// so you should replace the internal values of `data` with freshly allocated -// results to return different rows. -func (r *mockResults[T]) AddCurrentData() *mockResults[T] { - r.t.Helper() - - r.data.Reset() - err := r.tmpl.Execute(io.Discard, r.data) - require.NoError(r.t, err) - - d := r.data.GetScanDest() - dv := make([]driver.Value, len(d)) - for i, v := range d { - dv[i] = v - } - r.rows.AddRow(dv...) - - return r -} - -// Rows returns the *sqlmock.Rows object built. -func (r *mockResults[T]) Rows() *sqlmock.Rows { - return r.rows -} - -func TestPtrOr(t *testing.T) { - t.Parallel() - - p := ptrOr[*int]() - require.NotNil(t, p) - require.Zero(t, *p) - - p = ptrOr[*int](nil, nil, nil, nil, nil, nil) - require.NotNil(t, p) - require.Zero(t, *p) - - v := 42 - v2 := 5 - p = ptrOr(nil, nil, nil, &v, nil, &v2, nil, nil) - require.NotNil(t, p) - require.Equal(t, v, *p) - - p = ptrOr(nil, nil, nil, &v) - require.NotNil(t, p) - require.Equal(t, v, *p) -} - -func TestSliceOr(t *testing.T) { - t.Parallel() - - p := sliceOr[[]int]() - require.NotNil(t, p) - require.Len(t, p, 0) - - p = sliceOr[[]int](nil, nil, nil, nil) - require.NotNil(t, p) - require.Len(t, p, 0) - - p = sliceOr([]int{}, []int{}, []int{}, []int{}) - require.NotNil(t, p) - require.Len(t, p, 0) - - v := []int{1, 2} - p = sliceOr([]int{}, nil, []int{}, v, nil, []int{}, []int{10}, nil) - require.NotNil(t, p) - require.Equal(t, v, p) - - p = sliceOr([]int{}, nil, []int{}, v) - require.NotNil(t, p) - require.Equal(t, v, p) -} - -func TestMapOr(t *testing.T) { - t.Parallel() - - p := mapOr[map[string]int]() - require.NotNil(t, p) - require.Len(t, p, 0) - - p = mapOr(nil, map[string]int(nil), nil, map[string]int{}, nil) - require.NotNil(t, p) - require.Len(t, p, 0) - - v := map[string]int{"a": 0, "b": 1} - v2 := map[string]int{"c": 2, "d": 3} - - p = mapOr(nil, map[string]int(nil), v, v2, nil, map[string]int{}, nil) - require.NotNil(t, p) - require.Equal(t, v, p) - - p = mapOr(nil, map[string]int(nil), v) - require.NotNil(t, p) - require.Equal(t, v, p) -} - -var ( - validTestTmpl = template.Must(template.New("test").Parse("nothing special")) - invalidTestTmpl = template.New("no definition should fail to exec") - errTest = errors.New("because of reasons") -) - -// expectRows is a testing helper to keep mocks in sync when adding rows to a -// mocked SQL result. This is a helper to test `query` and `queryRow`. -type expectRows[T any] struct { - *sqlmock.Rows - ExpectedResults []T - - req *sqltemplateMocks.WithResults[T] -} - -func newReturnsRow[T any](dbmock sqlmock.Sqlmock, req *sqltemplateMocks.WithResults[T]) *expectRows[T] { - return &expectRows[T]{ - Rows: dbmock.NewRows(nil), - req: req, - } -} - -// Add adds a new value that should be returned by the `query` or `queryRow` -// operation. -func (r *expectRows[T]) Add(value T, err error) *expectRows[T] { - r.req.EXPECT().GetScanDest().Return(nil).Once() - r.req.EXPECT().Results().Return(value, err).Once() - r.Rows.AddRow() - r.ExpectedResults = append(r.ExpectedResults, value) - - return r -} - -func TestQueryRow(t *testing.T) { - t.Parallel() - - t.Run("happy path", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewWithResults[int64](t) - db, dbmock := newMockDBNopSQL(t) - rows := newReturnsRow(dbmock, req) - - // setup expectations - req.EXPECT().Validate().Return(nil).Once() - req.EXPECT().GetArgs().Return(nil).Once() - rows.Add(1, nil) - dbmock.ExpectQuery("").WillReturnRows(rows.Rows) - - // execute and assert - res, err := queryRow(ctx, db, validTestTmpl, req) - require.NoError(t, err) - require.Equal(t, rows.ExpectedResults[0], res) - }) - - t.Run("invalid request", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewWithResults[int64](t) - db, _ := newMockDBNopSQL(t) - - // setup expectations - req.EXPECT().Validate().Return(errTest).Once() - - // execute and assert - res, err := queryRow(ctx, db, invalidTestTmpl, req) - require.Zero(t, res) - require.Error(t, err) - require.ErrorContains(t, err, "invalid request") - }) - - t.Run("error executing template", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewWithResults[int64](t) - db, _ := newMockDBNopSQL(t) - - // setup expectations - req.EXPECT().Validate().Return(nil).Once() - - // execute and assert - res, err := queryRow(ctx, db, invalidTestTmpl, req) - require.Zero(t, res) - require.Error(t, err) - require.ErrorContains(t, err, "execute template") - }) - - t.Run("error executing query", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewWithResults[int64](t) - db, dbmock := newMockDBNopSQL(t) - - // setup expectations - req.EXPECT().Validate().Return(nil).Once() - req.EXPECT().GetArgs().Return(nil) - req.EXPECT().GetScanDest().Return(nil).Maybe() - dbmock.ExpectQuery("").WillReturnError(errTest) - - // execute and assert - res, err := queryRow(ctx, db, validTestTmpl, req) - require.Zero(t, res) - require.Error(t, err) - require.ErrorAs(t, err, new(SQLError)) - }) -} - -// scannerFunc is an adapter for the `scanner` interface. -type scannerFunc func(dest ...any) error - -func (f scannerFunc) Scan(dest ...any) error { - return f(dest...) -} - -func TestScanRow(t *testing.T) { - t.Parallel() - - const value int64 = 1 - - t.Run("happy path", func(t *testing.T) { - t.Parallel() - - // test declarations - req := sqltemplateMocks.NewWithResults[int64](t) - sc := scannerFunc(func(dest ...any) error { - return nil - }) - - // setup expectations - req.EXPECT().GetScanDest().Return(nil).Once() - req.EXPECT().Results().Return(value, nil).Once() - - // execute and assert - res, err := scanRow(sc, req) - require.NoError(t, err) - require.Equal(t, value, res) - }) - - t.Run("scan error", func(t *testing.T) { - t.Parallel() - - // test declarations - req := sqltemplateMocks.NewWithResults[int64](t) - sc := scannerFunc(func(dest ...any) error { - return errTest - }) - - // setup expectations - req.EXPECT().GetScanDest().Return(nil).Once() - - // execute and assert - res, err := scanRow(sc, req) - require.Zero(t, res) - require.Error(t, err) - require.ErrorIs(t, err, errTest) - }) - - t.Run("results error", func(t *testing.T) { - t.Parallel() - - // test declarations - req := sqltemplateMocks.NewWithResults[int64](t) - sc := scannerFunc(func(dest ...any) error { - return nil - }) - - // setup expectations - req.EXPECT().GetScanDest().Return(nil).Once() - req.EXPECT().Results().Return(0, errTest).Once() - - // execute and assert - res, err := scanRow(sc, req) - require.Zero(t, res) - require.Error(t, err) - require.ErrorIs(t, err, errTest) - }) -} - -func TestExec(t *testing.T) { - t.Parallel() - - t.Run("happy path", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewSQLTemplateIface(t) - db, dbmock := newMockDBNopSQL(t) - - // setup expectations - req.EXPECT().Validate().Return(nil).Once() - req.EXPECT().GetArgs().Return(nil).Once() - dbmock.ExpectExec("").WillReturnResult(sqlmock.NewResult(0, 0)) - - // execute and assert - res, err := exec(ctx, db, validTestTmpl, req) - require.NoError(t, err) - require.NotNil(t, res) - }) - - t.Run("invalid request", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewSQLTemplateIface(t) - db, _ := newMockDBNopSQL(t) - - // setup expectations - req.EXPECT().Validate().Return(errTest).Once() - - // execute and assert - res, err := exec(ctx, db, invalidTestTmpl, req) - require.Nil(t, res) - require.Error(t, err) - require.ErrorContains(t, err, "invalid request") - }) - - t.Run("error executing template", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewSQLTemplateIface(t) - db, _ := newMockDBNopSQL(t) - - // setup expectations - req.EXPECT().Validate().Return(nil).Once() - - // execute and assert - res, err := exec(ctx, db, invalidTestTmpl, req) - require.Nil(t, res) - require.Error(t, err) - require.ErrorContains(t, err, "execute template") - }) - - t.Run("error executing SQL", func(t *testing.T) { - t.Parallel() - - // test declarations - ctx := testutil.NewDefaultTestContext(t) - req := sqltemplateMocks.NewSQLTemplateIface(t) - db, dbmock := newMockDBNopSQL(t) - - // setup expectations - req.EXPECT().Validate().Return(nil).Once() - req.EXPECT().GetArgs().Return(nil) - dbmock.ExpectExec("").WillReturnError(errTest) - - // execute and assert - res, err := exec(ctx, db, validTestTmpl, req) - require.Nil(t, res) - require.Error(t, err) - require.ErrorAs(t, err, new(SQLError)) - }) -}