diff --git a/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go b/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go index a884c8f16cf..496adaf060d 100644 --- a/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go +++ b/pkg/cmd/grafana-cli/commands/datamigrations/to_unified_storage.go @@ -86,22 +86,30 @@ func ToUnifiedStorage(c utils.CommandLine, cfg *setting.Cfg, sqlStore db.DB) err ) if c.Bool("non-interactive") { - migrator := migrations.ProvideUnifiedMigrator(dashboardAccess, grpcClient) - - opts.WithHistory = true // always include history in non-interactive mode - rsp, err := migrator.Migrate(ctx, opts) - if exitErr := handleMigrationError(err, rsp); exitErr != nil { - return exitErr - } - - logger.Info("Migrated legacy resources successfully in", time.Since(start)) - if rsp != nil { - jj, _ := json.MarshalIndent(rsp, "", " ") - logger.Info("Migration summary:", string(jj)) - } - return nil + return runNonInteractiveMigration(ctx, opts, dashboardAccess, grpcClient, start) } + 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 { + migrator := migrations.ProvideUnifiedMigrator(dashboardAccess, grpcClient) + + opts.WithHistory = true // always include history in non-interactive mode + rsp, err := migrator.Migrate(ctx, opts) + if exitErr := handleMigrationError(err, rsp); exitErr != nil { + return exitErr + } + + logger.Info("Migrated legacy resources successfully in", time.Since(start)) + if rsp != nil { + jj, _ := json.MarshalIndent(rsp, "", " ") + logger.Info("Migration summary:", string(jj)) + } + return nil +} + +func runInteractiveMigration(ctx context.Context, cfg *setting.Cfg, opts legacy.MigrateOptions, dashboardAccess legacy.MigrationDashboardAccessor, grpcClient resource.ResourceClient, start time.Time) error { yes, err := promptYesNo(fmt.Sprintf("Count legacy resources for namespace: %s?", opts.Namespace)) if err != nil { return err @@ -143,7 +151,6 @@ func ToUnifiedStorage(c utils.CommandLine, cfg *setting.Cfg, sqlStore db.DB) err } migrator := migrations.ProvideUnifiedMigratorParquet(dashboardAccess, parquetClient) start = time.Now() - last = time.Now() rsp, err := migrator.Migrate(ctx, opts) if err != nil { return err @@ -187,7 +194,6 @@ func ToUnifiedStorage(c utils.CommandLine, cfg *setting.Cfg, sqlStore db.DB) err if yes { migrator := migrations.ProvideUnifiedMigrator(dashboardAccess, grpcClient) start = time.Now() - last = time.Now() rsp, err := migrator.Migrate(ctx, opts) if err != nil { return err diff --git a/pkg/registry/apis/dashboard/legacy/migration_dashboard_accessor_mock.go b/pkg/registry/apis/dashboard/legacy/migration_dashboard_accessor_mock.go index 43f2ce05ad1..dab60c6e29d 100644 --- a/pkg/registry/apis/dashboard/legacy/migration_dashboard_accessor_mock.go +++ b/pkg/registry/apis/dashboard/legacy/migration_dashboard_accessor_mock.go @@ -264,6 +264,67 @@ func (_c *MockMigrationDashboardAccessor_MigrateLibraryPanels_Call) RunAndReturn return _c } +// MigratePlaylists provides a mock function with given fields: ctx, orgId, opts, stream +func (_m *MockMigrationDashboardAccessor) MigratePlaylists(ctx context.Context, orgId int64, opts MigrateOptions, stream resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error) { + ret := _m.Called(ctx, orgId, opts, stream) + + if len(ret) == 0 { + panic("no return value specified for MigratePlaylists") + } + + var r0 *BlobStoreInfo + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, int64, MigrateOptions, resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error)); ok { + return rf(ctx, orgId, opts, stream) + } + if rf, ok := ret.Get(0).(func(context.Context, int64, MigrateOptions, resourcepb.BulkStore_BulkProcessClient) *BlobStoreInfo); ok { + r0 = rf(ctx, orgId, opts, stream) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*BlobStoreInfo) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, int64, MigrateOptions, resourcepb.BulkStore_BulkProcessClient) error); ok { + r1 = rf(ctx, orgId, opts, stream) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MockMigrationDashboardAccessor_MigratePlaylists_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'MigratePlaylists' +type MockMigrationDashboardAccessor_MigratePlaylists_Call struct { + *mock.Call +} + +// MigratePlaylists is a helper method to define mock.On call +// - ctx context.Context +// - orgId int64 +// - opts MigrateOptions +// - stream resourcepb.BulkStore_BulkProcessClient +func (_e *MockMigrationDashboardAccessor_Expecter) MigratePlaylists(ctx interface{}, orgId interface{}, opts interface{}, stream interface{}) *MockMigrationDashboardAccessor_MigratePlaylists_Call { + return &MockMigrationDashboardAccessor_MigratePlaylists_Call{Call: _e.mock.On("MigratePlaylists", ctx, orgId, opts, stream)} +} + +func (_c *MockMigrationDashboardAccessor_MigratePlaylists_Call) Run(run func(ctx context.Context, orgId int64, opts MigrateOptions, stream resourcepb.BulkStore_BulkProcessClient)) *MockMigrationDashboardAccessor_MigratePlaylists_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(int64), args[2].(MigrateOptions), args[3].(resourcepb.BulkStore_BulkProcessClient)) + }) + return _c +} + +func (_c *MockMigrationDashboardAccessor_MigratePlaylists_Call) Return(_a0 *BlobStoreInfo, _a1 error) *MockMigrationDashboardAccessor_MigratePlaylists_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *MockMigrationDashboardAccessor_MigratePlaylists_Call) RunAndReturn(run func(context.Context, int64, MigrateOptions, resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error)) *MockMigrationDashboardAccessor_MigratePlaylists_Call { + _c.Call.Return(run) + return _c +} + // NewMockMigrationDashboardAccessor creates a new instance of MockMigrationDashboardAccessor. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. // The first argument is typically a *testing.T value. func NewMockMigrationDashboardAccessor(t interface { diff --git a/pkg/registry/apis/dashboard/legacy/queries.go b/pkg/registry/apis/dashboard/legacy/queries.go index 56807b9cc5a..28be4f507a0 100644 --- a/pkg/registry/apis/dashboard/legacy/queries.go +++ b/pkg/registry/apis/dashboard/legacy/queries.go @@ -28,6 +28,7 @@ func mustTemplate(filename string) *template.Template { var ( sqlQueryDashboards = mustTemplate("query_dashboards.sql") sqlQueryPanels = mustTemplate("query_panels.sql") + sqlQueryPlaylists = mustTemplate("query_playlists.sql") ) type sqlQuery struct { @@ -83,3 +84,29 @@ func newLibraryQueryReq(sql *legacysql.LegacyDatabaseHelper, query *LibraryPanel UserTable: sql.Table("user"), } } + +type PlaylistQuery struct { + OrgID int64 +} + +type sqlPlaylistQuery struct { + sqltemplate.SQLTemplate + Query *PlaylistQuery + + PlaylistTable string + PlaylistItemTable string +} + +func (r sqlPlaylistQuery) Validate() error { + return nil +} + +func newPlaylistQueryReq(sql *legacysql.LegacyDatabaseHelper, query *PlaylistQuery) sqlPlaylistQuery { + return sqlPlaylistQuery{ + SQLTemplate: sqltemplate.New(sql.DialectForDriver()), + Query: query, + + PlaylistTable: sql.Table("playlist"), + PlaylistItemTable: sql.Table("playlist_item"), + } +} diff --git a/pkg/registry/apis/dashboard/legacy/queries_test.go b/pkg/registry/apis/dashboard/legacy/queries_test.go index 3ee82e899d7..1345bfc2925 100644 --- a/pkg/registry/apis/dashboard/legacy/queries_test.go +++ b/pkg/registry/apis/dashboard/legacy/queries_test.go @@ -29,6 +29,12 @@ func TestDashboardQueries(t *testing.T) { return &v } + getPlaylistQuery := func(q *PlaylistQuery) sqltemplate.SQLTemplate { + v := newPlaylistQueryReq(nodb, q) + v.SQLTemplate = mocks.NewTestingSQLTemplate() + return &v + } + mocks.CheckQuerySnapshots(t, mocks.TemplateTestSetup{ RootDir: "testdata", SQLTemplatesFS: sqlTemplatesFS, @@ -118,6 +124,14 @@ func TestDashboardQueries(t *testing.T) { }), }, }, + sqlQueryPlaylists: { + { + Name: "list", + Data: getPlaylistQuery(&PlaylistQuery{ + OrgID: 1, + }), + }, + }, }, }) } diff --git a/pkg/registry/apis/dashboard/legacy/query_playlists.sql b/pkg/registry/apis/dashboard/legacy/query_playlists.sql new file mode 100644 index 00000000000..5d37c05ad0c --- /dev/null +++ b/pkg/registry/apis/dashboard/legacy/query_playlists.sql @@ -0,0 +1,18 @@ +SELECT + p.id, + p.org_id, + p.uid, + p.name, + p.interval, + p.created_at, + p.updated_at, + pi.type as item_type, + pi.value as item_value +FROM + {{ .Ident .PlaylistTable }} as p + LEFT OUTER JOIN {{ .Ident .PlaylistItemTable }} as pi ON p.id = pi.playlist_id +WHERE + p.org_id = {{ .Arg .Query.OrgID }} +ORDER BY + p.id ASC, + pi.{{ .Ident "order" }} ASC \ No newline at end of file diff --git a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go index 8c465ef7b24..24ec135e785 100644 --- a/pkg/registry/apis/dashboard/legacy/sql_dashboards.go +++ b/pkg/registry/apis/dashboard/legacy/sql_dashboards.go @@ -24,6 +24,7 @@ import ( dashboardV1 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1beta1" "github.com/grafana/grafana/apps/dashboard/pkg/migration/schemaversion" folders "github.com/grafana/grafana/apps/folder/pkg/apis/folder/v1beta1" + playlistv0 "github.com/grafana/grafana/apps/playlist/pkg/apis/playlist/v0alpha1" "github.com/grafana/grafana/pkg/apimachinery/apis/common/v0alpha1" "github.com/grafana/grafana/pkg/apimachinery/identity" "github.com/grafana/grafana/pkg/apimachinery/utils" @@ -138,7 +139,16 @@ func NewDashboardSQLAccess(sql legacysql.LegacyDatabaseProvider, } } -func (a *dashboardSqlAccess) getRows(ctx context.Context, sql *legacysql.LegacyDatabaseHelper, query *DashboardQuery) (*rowsWrapper, error) { +func (a *dashboardSqlAccess) executeQuery(ctx context.Context, helper *legacysql.LegacyDatabaseHelper, query string, args ...any) (*sql.Rows, error) { + // Use transaction if available in context. + // This allows us to run migrations in a transaction which is specifically required for SQLite. + if tx := resource.TransactionFromContext(ctx); tx != nil { + return tx.QueryContext(ctx, query, args...) + } + return helper.DB.GetSqlxSession().Query(ctx, query, args...) +} + +func (a *dashboardSqlAccess) getRows(ctx context.Context, helper *legacysql.LegacyDatabaseHelper, query *DashboardQuery) (*rowsWrapper, error) { ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.getRows") defer span.End() @@ -150,7 +160,7 @@ func (a *dashboardSqlAccess) getRows(ctx context.Context, sql *legacysql.LegacyD // } } - req := newQueryReq(sql, query) + req := newQueryReq(helper, query) tmpl := sqlQueryDashboards if query.UseHistoryTable() && query.GetTrash { @@ -167,7 +177,7 @@ func (a *dashboardSqlAccess) getRows(ctx context.Context, sql *legacysql.LegacyD // fmt.Printf("DASHBOARD QUERY: %s [%+v] // %+v\n", pretty, req.GetArgs(), query) // } - rows, err := sql.DB.GetSqlxSession().Query(ctx, q, req.GetArgs()...) + rows, err := a.executeQuery(ctx, helper, q, req.GetArgs()...) if err != nil { if rows != nil { _ = rows.Close() @@ -465,6 +475,132 @@ func (a *dashboardSqlAccess) MigrateLibraryPanels(ctx context.Context, orgId int return nil, nil } +// MigratePlaylists handles the playlist migration logic +func (a *dashboardSqlAccess) MigratePlaylists(ctx context.Context, orgId int64, opts MigrateOptions, stream resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error) { + opts.Progress(-1, "migrating playlists...") + rows, err := a.ListPlaylists(ctx, orgId) + if rows != nil { + defer func() { + _ = rows.Close() + }() + } + if err != nil { + return nil, err + } + + // Group playlist items by playlist ID + type playlistData struct { + id int64 + uid string + name string + interval string + items []playlistv0.PlaylistItem + createdAt int64 + updatedAt int64 + } + + playlists := make(map[int64]*playlistData) + var currentID int64 + var orgID int64 + var uid, name, interval string + var createdAt, updatedAt int64 + var itemType, itemValue sql.NullString + + count := 0 + for rows.Next() { + err = rows.Scan(¤tID, &orgID, &uid, &name, &interval, &createdAt, &updatedAt, &itemType, &itemValue) + if err != nil { + return nil, err + } + + // Get or create playlist entry + pl, exists := playlists[currentID] + if !exists { + pl = &playlistData{ + id: currentID, + uid: uid, + name: name, + interval: interval, + items: []playlistv0.PlaylistItem{}, + createdAt: createdAt, + updatedAt: updatedAt, + } + playlists[currentID] = pl + } + + // Add item if it exists (LEFT JOIN can return NULL for playlists without items) + if itemType.Valid && itemValue.Valid { + pl.items = append(pl.items, playlistv0.PlaylistItem{ + Type: playlistv0.PlaylistItemType(itemType.String), + Value: itemValue.String, + }) + } + } + + if err = rows.Err(); err != nil { + return nil, err + } + + // Convert to K8s objects and send to stream + for _, pl := range playlists { + playlist := &playlistv0.Playlist{ + TypeMeta: metav1.TypeMeta{ + APIVersion: playlistv0.GroupVersion.String(), + Kind: "Playlist", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: pl.uid, + Namespace: opts.Namespace, + CreationTimestamp: metav1.NewTime(time.UnixMilli(pl.createdAt)), + }, + Spec: playlistv0.PlaylistSpec{ + Title: pl.name, + Interval: pl.interval, + Items: pl.items, + }, + } + + // Set updated timestamp if different from created + if pl.updatedAt != pl.createdAt { + meta, err := utils.MetaAccessor(playlist) + if err != nil { + return nil, err + } + updatedTime := time.UnixMilli(pl.updatedAt) + meta.SetUpdatedTimestamp(&updatedTime) + } + + body, err := json.Marshal(playlist) + if err != nil { + return nil, err + } + + req := &resourcepb.BulkRequest{ + Key: &resourcepb.ResourceKey{ + Namespace: opts.Namespace, + Group: "playlist.grafana.app", + Resource: "playlists", + Name: pl.uid, + }, + Value: body, + Action: resourcepb.BulkRequest_ADDED, + } + + opts.Progress(count, fmt.Sprintf("%s (%d)", pl.name, len(req.Value))) + count++ + + err = stream.Send(req) + if err != nil { + if errors.Is(err, io.EOF) { + err = nil + } + return nil, err + } + } + opts.Progress(-2, fmt.Sprintf("finished playlists... (%d)", len(playlists))) + return nil, nil +} + var _ resource.ListIterator = (*rowsWrapper)(nil) type rowsWrapper struct { @@ -903,20 +1039,19 @@ func (a *dashboardSqlAccess) GetLibraryPanels(ctx context.Context, query Library return nil, err } - sqlx, err := a.sql(ctx) + helper, err := a.sql(ctx) if err != nil { return nil, err } - req := newLibraryQueryReq(sqlx, &query) + req := newLibraryQueryReq(helper, &query) rawQuery, err := sqltemplate.Execute(sqlQueryPanels, req) if err != nil { return nil, fmt.Errorf("execute template %q: %w", sqlQueryPanels.Name(), err) } - q := rawQuery res := &dashboardV0.LibraryPanelList{} - rows, err := sqlx.DB.GetSqlxSession().Query(ctx, q, req.GetArgs()...) + rows, err := a.executeQuery(ctx, helper, rawQuery, req.GetArgs()...) defer func() { if rows != nil { _ = rows.Close() @@ -959,7 +1094,7 @@ func (a *dashboardSqlAccess) GetLibraryPanels(ctx context.Context, query Library } } if query.UID == "" { - rv, err := sqlx.GetResourceVersion(ctx, "library_element", "updated") + rv, err := helper.GetResourceVersion(ctx, "library_element", "updated") if err == nil { res.ResourceVersion = strconv.FormatInt(rv*1000, 10) // convert to microseconds } @@ -1038,3 +1173,29 @@ func parseLibraryPanelRow(p panel) (dashboardV0.LibraryPanel, error) { func (b *dashboardSqlAccess) RebuildIndexes(ctx context.Context, req *resourcepb.RebuildIndexesRequest) (*resourcepb.RebuildIndexesResponse, error) { return nil, fmt.Errorf("not implemented") } + +func (a *dashboardSqlAccess) ListPlaylists(ctx context.Context, orgID int64) (*sql.Rows, error) { + ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ListPlaylists") + defer span.End() + + helper, err := a.sql(ctx) + if err != nil { + return nil, err + } + + req := newPlaylistQueryReq(helper, &PlaylistQuery{ + OrgID: orgID, + }) + + rawQuery, err := sqltemplate.Execute(sqlQueryPlaylists, req) + if err != nil { + return nil, fmt.Errorf("execute template %q: %w", sqlQueryPlaylists.Name(), err) + } + + rows, err := a.executeQuery(ctx, helper, rawQuery, req.GetArgs()...) + if err != nil && rows != nil { + _ = rows.Close() + return nil, err + } + return rows, err +} diff --git a/pkg/registry/apis/dashboard/legacy/testdata/mysql--query_playlists-list.sql b/pkg/registry/apis/dashboard/legacy/testdata/mysql--query_playlists-list.sql new file mode 100755 index 00000000000..d1fc170c8df --- /dev/null +++ b/pkg/registry/apis/dashboard/legacy/testdata/mysql--query_playlists-list.sql @@ -0,0 +1,18 @@ +SELECT + p.id, + p.org_id, + p.uid, + p.name, + p.interval, + p.created_at, + p.updated_at, + pi.type as item_type, + pi.value as item_value +FROM + `grafana`.`playlist` as p + LEFT OUTER JOIN `grafana`.`playlist_item` as pi ON p.id = pi.playlist_id +WHERE + p.org_id = 1 +ORDER BY + p.id ASC, + pi.`order` ASC diff --git a/pkg/registry/apis/dashboard/legacy/testdata/postgres--query_playlists-list.sql b/pkg/registry/apis/dashboard/legacy/testdata/postgres--query_playlists-list.sql new file mode 100755 index 00000000000..4927dcaa049 --- /dev/null +++ b/pkg/registry/apis/dashboard/legacy/testdata/postgres--query_playlists-list.sql @@ -0,0 +1,18 @@ +SELECT + p.id, + p.org_id, + p.uid, + p.name, + p.interval, + p.created_at, + p.updated_at, + pi.type as item_type, + pi.value as item_value +FROM + "grafana"."playlist" as p + LEFT OUTER JOIN "grafana"."playlist_item" as pi ON p.id = pi.playlist_id +WHERE + p.org_id = 1 +ORDER BY + p.id ASC, + pi."order" ASC diff --git a/pkg/registry/apis/dashboard/legacy/testdata/sqlite--query_playlists-list.sql b/pkg/registry/apis/dashboard/legacy/testdata/sqlite--query_playlists-list.sql new file mode 100755 index 00000000000..4927dcaa049 --- /dev/null +++ b/pkg/registry/apis/dashboard/legacy/testdata/sqlite--query_playlists-list.sql @@ -0,0 +1,18 @@ +SELECT + p.id, + p.org_id, + p.uid, + p.name, + p.interval, + p.created_at, + p.updated_at, + pi.type as item_type, + pi.value as item_value +FROM + "grafana"."playlist" as p + LEFT OUTER JOIN "grafana"."playlist_item" as pi ON p.id = pi.playlist_id +WHERE + p.org_id = 1 +ORDER BY + p.id ASC, + pi."order" ASC diff --git a/pkg/registry/apis/dashboard/legacy/types.go b/pkg/registry/apis/dashboard/legacy/types.go index 31d08af16d8..26cd043e8e5 100644 --- a/pkg/registry/apis/dashboard/legacy/types.go +++ b/pkg/registry/apis/dashboard/legacy/types.go @@ -74,4 +74,5 @@ type MigrationDashboardAccessor interface { MigrateDashboards(ctx context.Context, orgId int64, opts MigrateOptions, stream resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error) MigrateFolders(ctx context.Context, orgId int64, opts MigrateOptions, stream resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error) MigrateLibraryPanels(ctx context.Context, orgId int64, opts MigrateOptions, stream resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error) + MigratePlaylists(ctx context.Context, orgId int64, opts MigrateOptions, stream resourcepb.BulkStore_BulkProcessClient) (*BlobStoreInfo, error) } diff --git a/pkg/setting/setting_unified_storage.go b/pkg/setting/setting_unified_storage.go index 9d0f5843ad2..4f69daa64fd 100644 --- a/pkg/setting/setting_unified_storage.go +++ b/pkg/setting/setting_unified_storage.go @@ -8,9 +8,8 @@ import ( "github.com/grafana/grafana/pkg/util/osutil" ) -// nolint:unused -var migratedUnifiedResources = []string{ - //"playlists.playlist.grafana.app", +var MigratedUnifiedResources = []string{ + "playlists.playlist.grafana.app", "folders.folder.grafana.app", "dashboards.dashboard.grafana.app", } @@ -63,7 +62,7 @@ func (cfg *Cfg) setUnifiedStorageConfig() { if !cfg.DisableDataMigrations && cfg.getUnifiedStorageType() == "unified" { // Helper log to find instances running migrations in the future cfg.Logger.Info("Unified migration configs not yet enforced") - //cfg.enforceMigrationToUnifiedConfigs() // TODO: uncomment when ready for release + // cfg.enforceMigrationToUnifiedConfigs() // TODO: uncomment when ready for release } else { // Helper log to find instances disabling migration cfg.Logger.Info("Unified migration configs enforcement disabled", "storage_type", cfg.getUnifiedStorageType(), "disable_data_migrations", cfg.DisableDataMigrations) @@ -116,7 +115,7 @@ func (cfg *Cfg) enforceMigrationToUnifiedConfigs() { section.Key("enable_search").SetValue("true") cfg.EnableSearch = true } - for _, resource := range migratedUnifiedResources { + for _, resource := range MigratedUnifiedResources { cfg.Logger.Info("Enforcing mode 5 for resource in unified storage", "resource", resource) if oldCfg, ok := cfg.UnifiedStorage[resource]; ok { cfg.Logger.Info("Overriding unified storage config for migrated resource", "resource", resource, "old_config", oldCfg) diff --git a/pkg/storage/unified/migrations/migrator.go b/pkg/storage/unified/migrations/migrator.go index d95aa59ace7..d086845d181 100644 --- a/pkg/storage/unified/migrations/migrator.go +++ b/pkg/storage/unified/migrations/migrator.go @@ -10,8 +10,6 @@ import ( authlib "github.com/grafana/authlib/types" - v1beta1 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1beta1" - folders "github.com/grafana/grafana/apps/folder/pkg/apis/folder/v1beta1" "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/grafana/grafana/pkg/storage/unified/resourcepb" ) @@ -35,39 +33,26 @@ type streamProvider interface { createStream(ctx context.Context, opts legacy.MigrateOptions) (resourcepb.BulkStore_BulkProcessClient, error) } -// resourceClientStreamProvider creates streams using resource.ResourceClient -type resourceClientStreamProvider struct { - client resource.ResourceClient -} - -func (r *resourceClientStreamProvider) createStream(ctx context.Context, opts legacy.MigrateOptions) (resourcepb.BulkStore_BulkProcessClient, error) { - // Build collection settings for resource client +func buildCollectionSettings(opts legacy.MigrateOptions) resource.BulkSettings { settings := resource.BulkSettings{ RebuildCollection: true, SkipValidation: true, } for _, res := range opts.Resources { - switch fmt.Sprintf("%s/%s", res.Group, res.Resource) { - case "folder.grafana.app/folders": - settings.Collection = append(settings.Collection, &resourcepb.ResourceKey{ - Namespace: opts.Namespace, - Group: folders.GROUP, - Resource: folders.RESOURCE, - }) - case "dashboard.grafana.app/librarypanels": - settings.Collection = append(settings.Collection, &resourcepb.ResourceKey{ - Namespace: opts.Namespace, - Group: v1beta1.GROUP, - Resource: v1beta1.LIBRARY_PANEL_RESOURCE, - }) - case "dashboard.grafana.app/dashboards": - settings.Collection = append(settings.Collection, &resourcepb.ResourceKey{ - Namespace: opts.Namespace, - Group: v1beta1.GROUP, - Resource: v1beta1.DASHBOARD_RESOURCE, - }) + key := buildResourceKey(res.Group, res.Resource, opts.Namespace) + if key != nil { + settings.Collection = append(settings.Collection, key) } } + return settings +} + +type resourceClientStreamProvider struct { + client resource.ResourceClient +} + +func (r *resourceClientStreamProvider) createStream(ctx context.Context, opts legacy.MigrateOptions) (resourcepb.BulkStore_BulkProcessClient, error) { + settings := buildCollectionSettings(opts) ctx = metadata.NewOutgoingContext(ctx, settings.ToMD()) return r.client.BulkProcess(ctx) } @@ -78,33 +63,7 @@ type bulkStoreClientStreamProvider struct { } func (b *bulkStoreClientStreamProvider) createStream(ctx context.Context, opts legacy.MigrateOptions) (resourcepb.BulkStore_BulkProcessClient, error) { - // Build collection settings for resource client - settings := resource.BulkSettings{ - RebuildCollection: true, - SkipValidation: true, - } - for _, res := range opts.Resources { - switch fmt.Sprintf("%s/%s", res.Group, res.Resource) { - case "folder.grafana.app/folders": - settings.Collection = append(settings.Collection, &resourcepb.ResourceKey{ - Namespace: opts.Namespace, - Group: folders.GROUP, - Resource: folders.RESOURCE, - }) - case "dashboard.grafana.app/librarypanels": - settings.Collection = append(settings.Collection, &resourcepb.ResourceKey{ - Namespace: opts.Namespace, - Group: v1beta1.GROUP, - Resource: v1beta1.LIBRARY_PANEL_RESOURCE, - }) - case "dashboard.grafana.app/dashboards": - settings.Collection = append(settings.Collection, &resourcepb.ResourceKey{ - Namespace: opts.Namespace, - Group: v1beta1.GROUP, - Resource: v1beta1.DASHBOARD_RESOURCE, - }) - } - } + settings := buildCollectionSettings(opts) ctx = metadata.NewOutgoingContext(ctx, settings.ToMD()) return b.client.BulkProcess(ctx) } @@ -170,16 +129,11 @@ func (m *unifiedMigration) Migrate(ctx context.Context, opts legacy.MigrateOptio migratorFuncs := []migratorFunc{} for _, res := range opts.Resources { - switch fmt.Sprintf("%s/%s", res.Group, res.Resource) { - case "folder.grafana.app/folders": - migratorFuncs = append(migratorFuncs, m.MigrateFolders) - case "dashboard.grafana.app/librarypanels": - migratorFuncs = append(migratorFuncs, m.MigrateLibraryPanels) - case "dashboard.grafana.app/dashboards": - migratorFuncs = append(migratorFuncs, m.MigrateDashboards) - default: - return nil, fmt.Errorf("unsupported resource: %s", res) + fn := getMigratorFunc(m.MigrationDashboardAccessor, res.Group, res.Resource) + if fn == nil { + return nil, fmt.Errorf("unsupported resource: %s/%s", res.Group, res.Resource) } + migratorFuncs = append(migratorFuncs, fn) } // Execute migrations diff --git a/pkg/storage/unified/migrations/migrator_test.go b/pkg/storage/unified/migrations/migrator_test.go index e75ed26b9e3..92afe698a98 100644 --- a/pkg/storage/unified/migrations/migrator_test.go +++ b/pkg/storage/unified/migrations/migrator_test.go @@ -45,6 +45,7 @@ func TestIntegrationMigrations(t *testing.T) { migrationTestCases := []resourceMigratorTestCase{ newFoldersAndDashboardsTestCase(), + newPlaylistsTestCase(), } runMigrationTestSuite(t, migrationTestCases) diff --git a/pkg/storage/unified/migrations/playlists_test.go b/pkg/storage/unified/migrations/playlists_test.go new file mode 100644 index 00000000000..00f06759865 --- /dev/null +++ b/pkg/storage/unified/migrations/playlists_test.go @@ -0,0 +1,119 @@ +package migrations_test + +import ( + "context" + "testing" + + authlib "github.com/grafana/authlib/types" + "github.com/grafana/grafana/pkg/infra/tracing" + "github.com/grafana/grafana/pkg/services/playlist" + "github.com/grafana/grafana/pkg/services/playlist/playlistimpl" + "github.com/grafana/grafana/pkg/tests/apis" + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +// playlistsTestCase tests the "playlists" ResourceMigration +type playlistsTestCase struct { + playlistUIDs []string +} + +// newPlaylistsTestCase creates a test case for the playlists migrator +func newPlaylistsTestCase() resourceMigratorTestCase { + return &playlistsTestCase{ + playlistUIDs: []string{}, + } +} + +func (tc *playlistsTestCase) name() string { + return "playlists" +} + +func (tc *playlistsTestCase) resources() []schema.GroupVersionResource { + return []schema.GroupVersionResource{ + { + Group: "playlist.grafana.app", + Version: "v0alpha1", + Resource: "playlists", + }, + } +} + +func (tc *playlistsTestCase) setup(t *testing.T, helper *apis.K8sTestHelper) { + t.Helper() + + // Get playlist service from the test environment + // The service writes directly to SQL storage, which works in Mode0 + env := helper.GetEnv() + playlistSvc := playlistimpl.ProvideService(env.SQLStore, tracing.InitializeTracerForTest()) + + // Use a non-existent dashboard UID for testing + // This avoids interfering with other test cases + nonExistentDashboardUID := "non-existent-dashboard-uid" + + // Create playlist with dashboard UID items (pointing to non-existent dashboard) + playlist1UID := createTestPlaylist(t, playlistSvc, helper.Org1.OrgID, "Playlist with Dashboard UIDs", "5m", []playlist.PlaylistItem{ + {Type: "dashboard_by_uid", Value: nonExistentDashboardUID, Order: 1}, + }) + tc.playlistUIDs = append(tc.playlistUIDs, playlist1UID) + + // Create playlist with tag items + playlist2UID := createTestPlaylist(t, playlistSvc, helper.Org1.OrgID, "Playlist with Tags", "10m", []playlist.PlaylistItem{ + {Type: "dashboard_by_tag", Value: "test-tag", Order: 1}, + {Type: "dashboard_by_tag", Value: "another-tag", Order: 2}, + }) + tc.playlistUIDs = append(tc.playlistUIDs, playlist2UID) + + // Create playlist with mixed items + playlist3UID := createTestPlaylist(t, playlistSvc, helper.Org1.OrgID, "Playlist with Mixed Items", "15m", []playlist.PlaylistItem{ + {Type: "dashboard_by_uid", Value: nonExistentDashboardUID, Order: 1}, + {Type: "dashboard_by_tag", Value: "mixed-tag", Order: 2}, + }) + tc.playlistUIDs = append(tc.playlistUIDs, playlist3UID) +} + +func (tc *playlistsTestCase) verify(t *testing.T, helper *apis.K8sTestHelper, shouldExist bool) { + t.Helper() + + expectedPlaylistCount := 0 + if shouldExist { + expectedPlaylistCount = len(tc.playlistUIDs) + } + + orgID := helper.Org1.OrgID + namespace := authlib.OrgNamespaceFormatter(orgID) + + // Verify playlists + playlistCli := helper.GetResourceClient(apis.ResourceClientArgs{ + User: helper.Org1.Admin, + Namespace: namespace, + GVR: schema.GroupVersionResource{ + Group: "playlist.grafana.app", + Version: "v0alpha1", + Resource: "playlists", + }, + }) + + verifyResourceCount(t, playlistCli, expectedPlaylistCount) + for _, uid := range tc.playlistUIDs { + verifyResource(t, playlistCli, uid, shouldExist) + } +} + +func createTestPlaylist(t *testing.T, playlistSvc playlist.Service, orgID int64, name, interval string, items []playlist.PlaylistItem) string { + t.Helper() + + cmd := &playlist.CreatePlaylistCommand{ + Name: name, + Interval: interval, + Items: items, + OrgId: orgID, + } + + result, err := playlistSvc.Create(context.Background(), cmd) + require.NoError(t, err) + require.NotNil(t, result) + require.NotEmpty(t, result.UID) + + return result.UID +} diff --git a/pkg/storage/unified/migrations/resources.go b/pkg/storage/unified/migrations/resources.go new file mode 100644 index 00000000000..0a4dcc7766e --- /dev/null +++ b/pkg/storage/unified/migrations/resources.go @@ -0,0 +1,100 @@ +package migrations + +import ( + "fmt" + + v1beta1 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1beta1" + folders "github.com/grafana/grafana/apps/folder/pkg/apis/folder/v1beta1" + playlists "github.com/grafana/grafana/apps/playlist/pkg/apis/playlist/v0alpha1" + "github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy" + "github.com/grafana/grafana/pkg/setting" + "github.com/grafana/grafana/pkg/storage/unified/resourcepb" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +type ResourceDefinition struct { + GroupResource schema.GroupResource + MigratorFunc string // Name of the method: "MigrateFolders", "MigrateDashboards", etc. +} + +var registeredResources = []ResourceDefinition{ + { + GroupResource: schema.GroupResource{Group: folders.GROUP, Resource: folders.RESOURCE}, + MigratorFunc: "MigrateFolders", + }, + { + GroupResource: schema.GroupResource{Group: v1beta1.GROUP, Resource: v1beta1.LIBRARY_PANEL_RESOURCE}, + MigratorFunc: "MigrateLibraryPanels", + }, + { + GroupResource: schema.GroupResource{Group: v1beta1.GROUP, Resource: v1beta1.DASHBOARD_RESOURCE}, + MigratorFunc: "MigrateDashboards", + }, + { + GroupResource: schema.GroupResource{Group: playlists.APIGroup, Resource: "playlists"}, + MigratorFunc: "MigratePlaylists", + }, +} + +func getResourceDefinition(group, resource string) *ResourceDefinition { + for i := range registeredResources { + r := ®isteredResources[i] + if r.GroupResource.Group == group && r.GroupResource.Resource == resource { + return r + } + } + return nil +} + +func buildResourceKey(group, resource, namespace string) *resourcepb.ResourceKey { + def := getResourceDefinition(group, resource) + if def == nil { + return nil + } + return &resourcepb.ResourceKey{ + Namespace: namespace, + Group: def.GroupResource.Group, + Resource: def.GroupResource.Resource, + } +} + +func getMigratorFunc(accessor legacy.MigrationDashboardAccessor, group, resource string) migratorFunc { + def := getResourceDefinition(group, resource) + if def == nil { + return nil + } + + switch def.MigratorFunc { + case "MigrateFolders": + return accessor.MigrateFolders + case "MigrateLibraryPanels": + return accessor.MigrateLibraryPanels + case "MigrateDashboards": + return accessor.MigrateDashboards + case "MigratePlaylists": + return accessor.MigratePlaylists + default: + return nil + } +} + +func validateRegisteredResources() error { + registeredMap := make(map[string]bool) + for _, gr := range registeredResources { + key := fmt.Sprintf("%s.%s", gr.GroupResource.Resource, gr.GroupResource.Group) + registeredMap[key] = true + } + + var missing []string + for _, expected := range setting.MigratedUnifiedResources { + if !registeredMap[expected] { + missing = append(missing, expected) + } + } + + if len(missing) > 0 { + return fmt.Errorf("resources declared in setting.MigratedUnifiedResources are not registered for migration: %v", missing) + } + + return nil +} diff --git a/pkg/storage/unified/migrations/service.go b/pkg/storage/unified/migrations/service.go index fa1f62ca86a..1fda9b42593 100644 --- a/pkg/storage/unified/migrations/service.go +++ b/pkg/storage/unified/migrations/service.go @@ -85,18 +85,18 @@ func RegisterMigrations( logger.Warn("Failed to register migrator metrics", "error", err) } + if err := validateRegisteredResources(); err != nil { + return err + } + // Register resource migrations registerDashboardAndFolderMigration(mg, migrator, client) + registerPlaylistMigration(mg, migrator, client) // Run all registered migrations (blocking) sec := cfg.Raw.Section("database") - migrationLocking := sec.Key("migration_locking").MustBool(true) - if mg.Dialect.DriverName() == sqlstoremigrator.SQLite { - // disable migration locking for SQLite to avoid "database is locked" errors in the bulk operations - migrationLocking = false - } if err := mg.RunMigrations(ctx, - migrationLocking, + sec.Key("migration_locking").MustBool(true), sec.Key("locking_attempt_timeout_sec").MustInt()); err != nil { return fmt.Errorf("unified storage data migration failed: %w", err) } @@ -106,13 +106,13 @@ func RegisterMigrations( } func registerDashboardAndFolderMigration(mg *sqlstoremigrator.Migrator, migrator UnifiedMigrator, client resource.ResourceClient) { - folders := schema.GroupResource{Group: "folder.grafana.app", Resource: "folders"} - dashboards := schema.GroupResource{Group: "dashboard.grafana.app", Resource: "dashboards"} + foldersDef := getResourceDefinition("folder.grafana.app", "folders") + dashboardsDef := getResourceDefinition("dashboard.grafana.app", "dashboards") driverName := mg.Dialect.DriverName() folderCountValidator := NewCountValidator( client, - folders, + foldersDef.GroupResource, "dashboard", "org_id = ? and is_folder = true", driverName, @@ -120,19 +120,40 @@ func registerDashboardAndFolderMigration(mg *sqlstoremigrator.Migrator, migrator dashboardCountValidator := NewCountValidator( client, - dashboards, + dashboardsDef.GroupResource, "dashboard", "org_id = ? and is_folder = false", driverName, ) - folderTreeValidator := NewFolderTreeValidator(client, folders, driverName) + folderTreeValidator := NewFolderTreeValidator(client, foldersDef.GroupResource, driverName) dashboardsAndFolders := NewResourceMigration( migrator, - []schema.GroupResource{folders, dashboards}, + []schema.GroupResource{foldersDef.GroupResource, dashboardsDef.GroupResource}, "folders-dashboards", []Validator{folderCountValidator, dashboardCountValidator, folderTreeValidator}, ) mg.AddMigration("folders and dashboards migration", dashboardsAndFolders) } + +func registerPlaylistMigration(mg *sqlstoremigrator.Migrator, migrator UnifiedMigrator, client resource.ResourceClient) { + playlistsDef := getResourceDefinition("playlist.grafana.app", "playlists") + driverName := mg.Dialect.DriverName() + + playlistCountValidator := NewCountValidator( + client, + playlistsDef.GroupResource, + "playlist", + "org_id = ?", + driverName, + ) + + playlistsMigration := NewResourceMigration( + migrator, + []schema.GroupResource{playlistsDef.GroupResource}, + "playlists", + []Validator{playlistCountValidator}, + ) + mg.AddMigration("playlists migration", playlistsMigration) +}