From 0d4ad01b65f9ce85342c3e46a20cabece7d4b823 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mustafa=20Sencer=20=C3=96zcan?= <32759850+mustafasencer@users.noreply.github.com> Date: Wed, 19 Nov 2025 09:09:08 +0100 Subject: [PATCH] feat: add unified data migrations for dashboard and folders (#113853) --- pkg/registry/apis/dashboard/legacy/migrate.go | 7 +- .../backgroundsvcs/background_services.go | 3 + pkg/server/wire.go | 3 + pkg/server/wire_gen.go | 13 +- pkg/storage/unified/client.go | 9 +- .../migrations/dashboard_folder_migration.go | 93 ++++++++++ pkg/storage/unified/migrations/migrations.go | 108 ++++++++++++ pkg/storage/unified/migrations/migrator.go | 166 ++++++++++++++++++ 8 files changed, 385 insertions(+), 17 deletions(-) create mode 100644 pkg/storage/unified/migrations/dashboard_folder_migration.go create mode 100644 pkg/storage/unified/migrations/migrations.go create mode 100644 pkg/storage/unified/migrations/migrator.go diff --git a/pkg/registry/apis/dashboard/legacy/migrate.go b/pkg/registry/apis/dashboard/legacy/migrate.go index bdc47bf985a..e43a0fb7629 100644 --- a/pkg/registry/apis/dashboard/legacy/migrate.go +++ b/pkg/registry/apis/dashboard/legacy/migrate.go @@ -34,9 +34,8 @@ type MigrateOptions struct { LargeObjects apistore.LargeObjectSupport BlobStore resourcepb.BlobStoreClient Resources []schema.GroupResource - WithHistory bool // only applies to dashboards - OnlyCount bool // just count the values - StackID string // stack identifier for logging + WithHistory bool // only applies to dashboards + OnlyCount bool // just count the values Progress func(count int, msg string) } @@ -140,7 +139,6 @@ func (a *dashboardSqlAccess) Migrate(ctx context.Context, opts MigrateOptions) ( // Now run each migration blobStore := BlobStoreInfo{} - opts.StackID = fmt.Sprintf("%d", info.StackID) // Pass stack ID through options a.log.Info("start migrating legacy resources", "namespace", opts.Namespace, "orgId", info.OrgID, "stackId", info.StackID) for _, m := range migratorFuncs { blobs, err := m(ctx, info.OrgID, opts, stream) @@ -318,7 +316,6 @@ func (a *dashboardSqlAccess) migrateDashboards(ctx context.Context, orgId int64, "uid", row.Dash.UID, "id", id, "version", row.Dash.Generation, - "stackId", opts.StackID, ) opts.Progress(-2, fmt.Sprintf("rejected: id:%s, uid:%s", id, row.Dash.Name)) } diff --git a/pkg/registry/backgroundsvcs/background_services.go b/pkg/registry/backgroundsvcs/background_services.go index e80bcb720c9..5f997875985 100644 --- a/pkg/registry/backgroundsvcs/background_services.go +++ b/pkg/registry/backgroundsvcs/background_services.go @@ -48,6 +48,7 @@ import ( "github.com/grafana/grafana/pkg/services/supportbundles/supportbundlesimpl" "github.com/grafana/grafana/pkg/services/team/teamapi" "github.com/grafana/grafana/pkg/services/updatemanager" + unifiedmigrations "github.com/grafana/grafana/pkg/storage/unified/migrations" ) func ProvideBackgroundServiceRegistry( @@ -73,6 +74,7 @@ func ProvideBackgroundServiceRegistry( dashboardServiceImpl *service.DashboardServiceImpl, secretsGarbageCollectionWorker *secretsgarbagecollectionworker.Worker, fixedRolesLoader *accesscontrol.FixedRolesLoader, + unifiedStorageMigrationProvider unifiedmigrations.UnifiedStorageMigrationProvider, // Need to make sure these are initialized, is there a better place to put them? _ dashboardsnapshots.Service, _ serviceaccounts.Service, @@ -89,6 +91,7 @@ func ProvideBackgroundServiceRegistry( notifications, rendering, tokenService, + unifiedStorageMigrationProvider, provisioning, grafanaUpdateChecker, pluginsUpdateChecker, diff --git a/pkg/server/wire.go b/pkg/server/wire.go index 179b371760d..2fe6db16b31 100644 --- a/pkg/server/wire.go +++ b/pkg/server/wire.go @@ -181,6 +181,7 @@ import ( secretencryption "github.com/grafana/grafana/pkg/storage/secret/encryption" secretmetadata "github.com/grafana/grafana/pkg/storage/secret/metadata" secretmigrator "github.com/grafana/grafana/pkg/storage/secret/migrator" + unifiedmigrations "github.com/grafana/grafana/pkg/storage/unified/migrations" "github.com/grafana/grafana/pkg/storage/unified/resource" unifiedsearch "github.com/grafana/grafana/pkg/storage/unified/search" "github.com/grafana/grafana/pkg/tsdb/azuremonitor" @@ -465,6 +466,8 @@ var wireBasicSet = wire.NewSet( // Unified storage resource.ProvideStorageMetrics, resource.ProvideIndexMetrics, + unifiedmigrations.ProvideUnifiedStorageMigrationProvider, + wire.Bind(new(unifiedmigrations.UnifiedStorageMigrationProvider), new(*unifiedmigrations.UnifiedStorageMigrationProviderImpl)), // Kubernetes API server grafanaapiserver.WireSet, apiregistry.WireSet, diff --git a/pkg/server/wire_gen.go b/pkg/server/wire_gen.go index 02b1db115b9..91a1e50a224 100644 --- a/pkg/server/wire_gen.go +++ b/pkg/server/wire_gen.go @@ -261,6 +261,7 @@ import ( "github.com/grafana/grafana/pkg/storage/secret/metadata" "github.com/grafana/grafana/pkg/storage/secret/migrator" "github.com/grafana/grafana/pkg/storage/unified" + migrations3 "github.com/grafana/grafana/pkg/storage/unified/migrations" "github.com/grafana/grafana/pkg/storage/unified/resource" "github.com/grafana/grafana/pkg/storage/unified/search" "github.com/grafana/grafana/pkg/storage/unified/sql" @@ -837,6 +838,8 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api dashboardUpdater := service8.ProvideDashboardUpdater(inProcBus, pluginstoreService, service14, importDashboardService, service13, pluginService, dashboardService) worker := garbagecollectionworker.ProvideWorker(cfg, secureValueMetadataStorage, keeperMetadataStorage, ossKeeperService) fixedRolesLoader := accesscontrol.ProvideFixedRolesLoader(acimplService, featureToggles) + legacyMigrator := legacy.ProvideLegacyMigrator(sqlStore, provisioningServiceImpl, libraryPanelService, dashboardPermissionsService, accessControl, featureToggles) + unifiedStorageMigrationProviderImpl := migrations3.ProvideUnifiedStorageMigrationProvider(legacyMigrator, cfg, resourceClient, sqlStore) healthService, err := grpcserver.ProvideHealthService(cfg, grpcserverProvider) if err != nil { return nil, err @@ -869,7 +872,6 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api userStorageAPIBuilder := userstorage.RegisterAPIService(featureToggles, apiserverService, registerer) apiBuilder := preferences.RegisterAPIService(cfg, featureToggles, sqlStore, prefService, userService, apiserverService) collectionsAPIBuilder := collections.RegisterAPIService(cfg, featureToggles, sqlStore, starService, userService, apiserverService) - legacyMigrator := legacy.ProvideLegacyMigrator(sqlStore, provisioningServiceImpl, libraryPanelService, dashboardPermissionsService, accessControl, featureToggles) webhookExtraBuilder := webhooks.ProvideWebhooksWithImages(cfg, renderingService, resourceClient, eventualRestConfigProvider, registerer) v3 := extras.ProvideProvisioningExtraAPIs(webhookExtraBuilder) pullRequestWorker := pullrequest.ProvidePullRequestWorker(cfg, renderingService, resourceClient, eventualRestConfigProvider, registerer) @@ -918,7 +920,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api } ossUserProtectionImpl := authinfoimpl.ProvideOSSUserProtectionService() registration := authnimpl.ProvideRegistration(cfg, authnService, orgService, userAuthTokenService, acimplService, permissionRegistry, apikeyService, userService, authService, ossUserProtectionImpl, loginattemptimplService, quotaService, authinfoimplService, renderingService, featureToggles, oauthtokenService, socialService, remoteCache, ldapImpl, ossImpl, tracingService, tempuserService, notificationService) - backgroundServiceRegistry := backgroundsvcs.ProvideBackgroundServiceRegistry(httpServer, alertNG, cleanUpService, grafanaLive, gateway, notificationService, pluginstoreService, renderingService, userAuthTokenService, tracingService, provisioningServiceImpl, usageStats, statscollectorService, grafanaService, pluginsService, internalMetricsService, secretsService, remoteCache, storageService, searchService, entityEventsService, serviceAccountsService, grpcserverProvider, secretMigrationProviderImpl, loginattemptimplService, supportbundlesimplService, metricService, keyRetriever, angulardetectorsproviderDynamic, apiserverService, anonDeviceService, ssosettingsimplService, pluginexternalService, plugininstallerService, zanzanaReconciler, appregistryService, dashboardUpdater, dashboardServiceImpl, worker, fixedRolesLoader, serviceImpl, serviceAccountsProxy, healthService, reflectionService, apiService, apiregistryService, idimplService, teamAPI, ssosettingsimplService, cloudmigrationService, registration) + backgroundServiceRegistry := backgroundsvcs.ProvideBackgroundServiceRegistry(httpServer, alertNG, cleanUpService, grafanaLive, gateway, notificationService, pluginstoreService, renderingService, userAuthTokenService, tracingService, provisioningServiceImpl, usageStats, statscollectorService, grafanaService, pluginsService, internalMetricsService, secretsService, remoteCache, storageService, searchService, entityEventsService, serviceAccountsService, grpcserverProvider, secretMigrationProviderImpl, loginattemptimplService, supportbundlesimplService, metricService, keyRetriever, angulardetectorsproviderDynamic, apiserverService, anonDeviceService, ssosettingsimplService, pluginexternalService, plugininstallerService, zanzanaReconciler, appregistryService, dashboardUpdater, dashboardServiceImpl, worker, fixedRolesLoader, unifiedStorageMigrationProviderImpl, serviceImpl, serviceAccountsProxy, healthService, reflectionService, apiService, apiregistryService, idimplService, teamAPI, ssosettingsimplService, cloudmigrationService, registration) usageStatsProvidersRegistry := usagestatssvcs.ProvideUsageStatsProvidersRegistry(acimplService, userService) server, err := New(opts, cfg, httpServer, acimplService, provisioningServiceImpl, backgroundServiceRegistry, usageStatsProvidersRegistry, statscollectorService, tracingService, featureToggles, registerer) if err != nil { @@ -1479,6 +1481,8 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac dashboardUpdater := service8.ProvideDashboardUpdater(inProcBus, pluginstoreService, service14, importDashboardService, service13, pluginService, dashboardService) worker := garbagecollectionworker.ProvideWorker(cfg, secureValueMetadataStorage, keeperMetadataStorage, ossKeeperService) fixedRolesLoader := accesscontrol.ProvideFixedRolesLoader(acimplService, featureToggles) + legacyMigrator := legacy.ProvideLegacyMigrator(sqlStore, provisioningServiceImpl, libraryPanelService, dashboardPermissionsService, accessControl, featureToggles) + unifiedStorageMigrationProviderImpl := migrations3.ProvideUnifiedStorageMigrationProvider(legacyMigrator, cfg, resourceClient, sqlStore) healthService, err := grpcserver.ProvideHealthService(cfg, grpcserverProvider) if err != nil { return nil, err @@ -1511,7 +1515,6 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac userStorageAPIBuilder := userstorage.RegisterAPIService(featureToggles, apiserverService, registerer) apiBuilder := preferences.RegisterAPIService(cfg, featureToggles, sqlStore, prefService, userService, apiserverService) collectionsAPIBuilder := collections.RegisterAPIService(cfg, featureToggles, sqlStore, starService, userService, apiserverService) - legacyMigrator := legacy.ProvideLegacyMigrator(sqlStore, provisioningServiceImpl, libraryPanelService, dashboardPermissionsService, accessControl, featureToggles) webhookExtraBuilder := webhooks.ProvideWebhooksWithImages(cfg, renderingService, resourceClient, eventualRestConfigProvider, registerer) v3 := extras.ProvideProvisioningExtraAPIs(webhookExtraBuilder) pullRequestWorker := pullrequest.ProvidePullRequestWorker(cfg, renderingService, resourceClient, eventualRestConfigProvider, registerer) @@ -1560,7 +1563,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac } ossUserProtectionImpl := authinfoimpl.ProvideOSSUserProtectionService() registration := authnimpl.ProvideRegistration(cfg, authnService, orgService, userAuthTokenService, acimplService, permissionRegistry, apikeyService, userService, authService, ossUserProtectionImpl, loginattemptimplService, quotaService, authinfoimplService, renderingService, featureToggles, oauthtokentestService, socialService, remoteCache, ldapImpl, ossImpl, tracingService, tempuserService, notificationServiceMock) - backgroundServiceRegistry := backgroundsvcs.ProvideBackgroundServiceRegistry(httpServer, alertNG, cleanUpService, grafanaLive, gateway, notificationService, pluginstoreService, renderingService, userAuthTokenService, tracingService, provisioningServiceImpl, usageStats, statscollectorService, grafanaService, pluginsService, internalMetricsService, secretsService, remoteCache, storageService, searchService, entityEventsService, serviceAccountsService, grpcserverProvider, secretMigrationProviderImpl, loginattemptimplService, supportbundlesimplService, metricService, keyRetriever, angulardetectorsproviderDynamic, apiserverService, anonDeviceService, ssosettingsimplService, pluginexternalService, plugininstallerService, zanzanaReconciler, appregistryService, dashboardUpdater, dashboardServiceImpl, worker, fixedRolesLoader, serviceImpl, serviceAccountsProxy, healthService, reflectionService, apiService, apiregistryService, idimplService, teamAPI, ssosettingsimplService, cloudmigrationService, registration) + backgroundServiceRegistry := backgroundsvcs.ProvideBackgroundServiceRegistry(httpServer, alertNG, cleanUpService, grafanaLive, gateway, notificationService, pluginstoreService, renderingService, userAuthTokenService, tracingService, provisioningServiceImpl, usageStats, statscollectorService, grafanaService, pluginsService, internalMetricsService, secretsService, remoteCache, storageService, searchService, entityEventsService, serviceAccountsService, grpcserverProvider, secretMigrationProviderImpl, loginattemptimplService, supportbundlesimplService, metricService, keyRetriever, angulardetectorsproviderDynamic, apiserverService, anonDeviceService, ssosettingsimplService, pluginexternalService, plugininstallerService, zanzanaReconciler, appregistryService, dashboardUpdater, dashboardServiceImpl, worker, fixedRolesLoader, unifiedStorageMigrationProviderImpl, serviceImpl, serviceAccountsProxy, healthService, reflectionService, apiService, apiregistryService, idimplService, teamAPI, ssosettingsimplService, cloudmigrationService, registration) usageStatsProvidersRegistry := usagestatssvcs.ProvideUsageStatsProvidersRegistry(acimplService, userService) server, err := New(opts, cfg, httpServer, acimplService, provisioningServiceImpl, backgroundServiceRegistry, usageStatsProvidersRegistry, statscollectorService, tracingService, featureToggles, registerer) if err != nil { @@ -1756,7 +1759,7 @@ var withOTelSet = wire.NewSet( otelTracer, grpcserver.ProvideService, interceptors.ProvideAuthenticator, ) -var wireBasicSet = wire.NewSet(annotationsimpl.ProvideService, wire.Bind(new(annotations.Repository), new(*annotationsimpl.RepositoryImpl)), New, api.ProvideHTTPServer, query.ProvideService, wire.Bind(new(query.Service), new(*query.ServiceImpl)), bus.ProvideBus, wire.Bind(new(bus.Bus), new(*bus.InProcBus)), rendering.ProvideService, wire.Bind(new(rendering.Service), new(*rendering.RenderingService)), routing.ProvideRegister, wire.Bind(new(routing.RouteRegister), new(*routing.RouteRegisterImpl)), hooks.ProvideService, kvstore.ProvideService, localcache.ProvideService, bundleregistry.ProvideService, wire.Bind(new(supportbundles.Service), new(*bundleregistry.Service)), updatemanager.ProvideGrafanaService, updatemanager.ProvidePluginsService, service.ProvideService, wire.Bind(new(usagestats.Service), new(*service.UsageStats)), validator3.ProvideService, legacy.ProvideLegacyMigrator, pluginsintegration.WireSet, dashboards.ProvideFileStoreManager, wire.Bind(new(dashboards.FileStore), new(*dashboards.FileStoreManager)), cloudwatch.ProvideService, cloudmonitoring.ProvideService, azuremonitor.ProvideService, postgres.ProvideService, mysql.ProvideService, mssql.ProvideService, store.ProvideEntityEventsService, dualwrite.ProvideService, httpclientprovider.New, wire.Bind(new(httpclient.Provider), new(*httpclient2.Provider)), serverlock.ProvideService, wire.Bind(new(installsync.ServerLock), new(*serverlock.ServerLockService)), annotationsimpl.ProvideCleanupService, wire.Bind(new(annotations.Cleaner), new(*annotationsimpl.CleanupServiceImpl)), cleanup.ProvideService, shorturlimpl.ProvideService, wire.Bind(new(shorturls.Service), new(*shorturlimpl.ShortURLService)), queryhistory.ProvideService, wire.Bind(new(queryhistory.Service), new(*queryhistory.QueryHistoryService)), correlations.ProvideService, wire.Bind(new(correlations.Service), new(*correlations.CorrelationsService)), quotaimpl.ProvideService, remotecache.ProvideService, wire.Bind(new(remotecache.CacheStorage), new(*remotecache.RemoteCache)), authinfoimpl.ProvideService, wire.Bind(new(login.AuthInfoService), new(*authinfoimpl.Service)), authinfoimpl.ProvideStore, datasourceproxy.ProvideService, sort.ProvideService, search2.ProvideService, searchV2.ProvideService, searchV2.ProvideSearchHTTPService, store.ProvideService, store.ProvideSystemUsersService, live.ProvideService, pushhttp.ProvideService, contexthandler.ProvideService, service12.ProvideService, wire.Bind(new(service12.LDAP), new(*service12.LDAPImpl)), jwt.ProvideService, wire.Bind(new(jwt.JWTService), new(*jwt.AuthService)), store2.ProvideDBStore, image.ProvideDeleteExpiredService, ngalert.ProvideService, librarypanels.ProvideService, wire.Bind(new(librarypanels.Service), new(*librarypanels.LibraryPanelService)), libraryelements.ProvideService, wire.Bind(new(libraryelements.Service), new(*libraryelements.LibraryElementService)), notifications.ProvideService, notifications.ProvideSmtpService, github.ProvideFactory, tracing.ProvideService, tracing.ProvideTracingConfig, wire.Bind(new(tracing.Tracer), new(*tracing.TracingService)), withOTelSet, testdatasource.ProvideService, api4.ProvideService, opentsdb.ProvideService, socialimpl.ProvideService, influxdb.ProvideService, wire.Bind(new(social.Service), new(*socialimpl.SocialService)), tempo.ProvideService, loki.ProvideService, graphite.ProvideService, prometheus.ProvideService, elasticsearch.ProvideService, pyroscope.ProvideService, parca.ProvideService, zipkin.ProvideService, jaeger.ProvideService, service9.ProvideCacheService, wire.Bind(new(datasources.CacheService), new(*service9.CacheServiceImpl)), service2.ProvideEncryptionService, wire.Bind(new(encryption2.Internal), new(*service2.Service)), manager.ProvideSecretsService, wire.Bind(new(secrets.Service), new(*manager.SecretsService)), database.ProvideSecretsStore, wire.Bind(new(secrets.Store), new(*database.SecretsStoreImpl)), garbagecollectionworker.ProvideWorker, grafanads.ProvideService, wire.Bind(new(dashboardsnapshots.Store), new(*database5.DashboardSnapshotStore)), database5.ProvideStore, wire.Bind(new(dashboardsnapshots.Service), new(*service10.ServiceImpl)), service10.ProvideService, service9.ProvideService, wire.Bind(new(datasources.DataSourceService), new(*service9.Service)), service9.ProvideLegacyDataSourceLookup, retriever.ProvideService, wire.Bind(new(serviceaccounts.ServiceAccountRetriever), new(*retriever.Service)), ossaccesscontrol.ProvideServiceAccountPermissions, wire.Bind(new(accesscontrol.ServiceAccountPermissionsService), new(*ossaccesscontrol.ServiceAccountPermissionsService)), manager3.ProvideServiceAccountsService, proxy.ProvideServiceAccountsProxy, wire.Bind(new(serviceaccounts.Service), new(*proxy.ServiceAccountsProxy)), dsquerierclient.NewNullQSDatasourceClientBuilder, expr.ProvideService, featuremgmt.ProvideManagerService, featuremgmt.ProvideToggles, service7.ProvideDashboardServiceImpl, wire.Bind(new(dashboards2.PermissionsRegistrationService), new(*service7.DashboardServiceImpl)), service7.ProvideDashboardService, service7.ProvideDashboardProvisioningService, service7.ProvideDashboardPluginService, database2.ProvideDashboardStore, folderimpl.ProvideService, wire.Bind(new(folder.Service), new(*folderimpl.Service)), wire.Bind(new(folder.LegacyService), new(*folderimpl.Service)), folderimpl.ProvideStore, wire.Bind(new(folder.Store), new(*folderimpl.FolderStoreImpl)), service11.ProvideService, wire.Bind(new(dashboardimport.Service), new(*service11.ImportDashboardService)), service8.ProvideService, wire.Bind(new(plugindashboards.Service), new(*service8.Service)), service8.ProvideDashboardUpdater, kvstore2.ProvideService, avatar.ProvideAvatarCacheServer, statscollector.ProvideService, csrf.ProvideCSRFFilter, wire.Bind(new(csrf.Service), new(*csrf.CSRF)), ossaccesscontrol.ProvideTeamPermissions, wire.Bind(new(accesscontrol.TeamPermissionsService), new(*ossaccesscontrol.TeamPermissionsService)), ossaccesscontrol.ProvideFolderPermissions, wire.Bind(new(accesscontrol.FolderPermissionsService), new(*ossaccesscontrol.FolderPermissionsService)), ossaccesscontrol.ProvideDashboardPermissions, wire.Bind(new(accesscontrol.DashboardPermissionsService), new(*ossaccesscontrol.DashboardPermissionsService)), ossaccesscontrol.ProvideReceiverPermissionsService, wire.Bind(new(accesscontrol.ReceiverPermissionsService), new(*ossaccesscontrol.ReceiverPermissionsService)), starimpl.ProvideService, playlistimpl.ProvideService, apikeyimpl.ProvideService, dashverimpl.ProvideService, service3.ProvideService, wire.Bind(new(publicdashboards.Service), new(*service3.PublicDashboardServiceImpl)), database3.ProvideStore, wire.Bind(new(publicdashboards.Store), new(*database3.PublicDashboardStoreImpl)), metric.ProvideService, api2.ProvideApi, api3.ProvideApi, userimpl.ProvideService, orgimpl.ProvideService, orgimpl.ProvideDeletionService, statsimpl.ProvideService, grpccontext.ProvideContextHandler, grpcserver.ProvideHealthService, grpcserver.ProvideReflectionService, resolver.ProvideEntityReferenceResolver, teamimpl.ProvideService, teamapi.ProvideTeamAPI, tempuserimpl.ProvideService, loginattemptimpl.ProvideService, wire.Bind(new(loginattempt.Service), new(*loginattemptimpl.Service)), migrations2.ProvideDataSourceMigrationService, migrations2.ProvideSecretMigrationProvider, wire.Bind(new(migrations2.SecretMigrationProvider), new(*migrations2.SecretMigrationProviderImpl)), promtypemigration.ProvideAzurePromMigrationService, promtypemigration.ProvideAmazonPromMigrationService, promtypemigration.ProvidePromTypeMigrationProvider, wire.Bind(new(promtypemigration.PromTypeMigrationProvider), new(*promtypemigration.PromTypeMigrationProviderImpl)), resourcepermissions.NewActionSetService, wire.Bind(new(accesscontrol.ActionResolver), new(resourcepermissions.ActionSetService)), wire.Bind(new(pluginaccesscontrol.ActionSetRegistry), new(resourcepermissions.ActionSetService)), permreg.ProvidePermissionRegistry, acimpl.ProvideAccessControl, accesscontrol.ProvideFixedRolesLoader, dualwrite2.ProvideZanzanaReconciler, navtreeimpl.ProvideService, wire.Bind(new(accesscontrol.AccessControl), new(*acimpl.AccessControl)), wire.Bind(new(notifications.TempUserStore), new(tempuser.Service)), tagimpl.ProvideService, wire.Bind(new(tag.Service), new(*tagimpl.Service)), authnimpl.ProvideService, authnimpl.ProvideIdentitySynchronizer, authnimpl.ProvideAuthnService, authnimpl.ProvideAuthnServiceAuthenticateOnly, authnimpl.ProvideRegistration, supportbundlesimpl.ProvideService, extsvcaccounts.ProvideExtSvcAccountsService, wire.Bind(new(serviceaccounts.ExtSvcAccountsService), new(*extsvcaccounts.ExtSvcAccountsService)), registry2.ProvideExtSvcRegistry, wire.Bind(new(extsvcauth.ExternalServiceRegistry), new(*registry2.Registry)), anonstore.ProvideAnonDBStore, wire.Bind(new(anonstore.AnonStore), new(*anonstore.AnonDBStore)), loggermw.Provide, slogadapter.Provide, signingkeysimpl.ProvideEmbeddedSigningKeysService, wire.Bind(new(signingkeys.Service), new(*signingkeysimpl.Service)), ssosettingsimpl.ProvideService, wire.Bind(new(ssosettings.Service), new(*ssosettingsimpl.Service)), idimpl.ProvideService, wire.Bind(new(auth.IDService), new(*idimpl.Service)), cloudmigrationimpl.ProvideService, caching.ProvideCachingServiceClient, userimpl.ProvideVerifier, connectors.ProvideOrgRoleMapper, wire.Bind(new(user.Verifier), new(*userimpl.Verifier)), authz.WireSet, metadata.ProvideSecureValueMetadataStorage, metadata.ProvideKeeperMetadataStorage, metadata.ProvideDecryptStorage, decrypt.ProvideDecryptAuthorizer, wire.Value([]decrypt.ExtraOwnerDecrypter(nil)), decrypt.ProvideDecryptService, inline.ProvideInlineSecureValueService, encryption.ProvideDataKeyStorage, encryption.ProvideGlobalDataKeyStorage, encryption.ProvideEncryptedValueStorage, encryption.ProvideGlobalEncryptedValueStorage, encryption.ProvideEncryptedValueMigrationExecutor, service5.ProvideSecureValueService, validator.ProvideKeeperValidator, validator.ProvideSecureValueValidator, mutator.ProvideKeeperMutator, mutator.ProvideSecureValueMutator, migrator.NewWithEngine, database4.ProvideDatabase, clock.ProvideClock, wire.Bind(new(contracts.Database), new(*database4.Database)), wire.Bind(new(contracts.Clock), new(*clock.Clock)), manager2.ProvideEncryptionManager, service4.ProvideAESGCMCipherService, resource.ProvideStorageMetrics, resource.ProvideIndexMetrics, apiserver.WireSet, apiregistry.WireSet, appregistry.WireSet, client.ProvideK8sClientWithFallback) +var wireBasicSet = wire.NewSet(annotationsimpl.ProvideService, wire.Bind(new(annotations.Repository), new(*annotationsimpl.RepositoryImpl)), New, api.ProvideHTTPServer, query.ProvideService, wire.Bind(new(query.Service), new(*query.ServiceImpl)), bus.ProvideBus, wire.Bind(new(bus.Bus), new(*bus.InProcBus)), rendering.ProvideService, wire.Bind(new(rendering.Service), new(*rendering.RenderingService)), routing.ProvideRegister, wire.Bind(new(routing.RouteRegister), new(*routing.RouteRegisterImpl)), hooks.ProvideService, kvstore.ProvideService, localcache.ProvideService, bundleregistry.ProvideService, wire.Bind(new(supportbundles.Service), new(*bundleregistry.Service)), updatemanager.ProvideGrafanaService, updatemanager.ProvidePluginsService, service.ProvideService, wire.Bind(new(usagestats.Service), new(*service.UsageStats)), validator3.ProvideService, legacy.ProvideLegacyMigrator, pluginsintegration.WireSet, dashboards.ProvideFileStoreManager, wire.Bind(new(dashboards.FileStore), new(*dashboards.FileStoreManager)), cloudwatch.ProvideService, cloudmonitoring.ProvideService, azuremonitor.ProvideService, postgres.ProvideService, mysql.ProvideService, mssql.ProvideService, store.ProvideEntityEventsService, dualwrite.ProvideService, httpclientprovider.New, wire.Bind(new(httpclient.Provider), new(*httpclient2.Provider)), serverlock.ProvideService, wire.Bind(new(installsync.ServerLock), new(*serverlock.ServerLockService)), annotationsimpl.ProvideCleanupService, wire.Bind(new(annotations.Cleaner), new(*annotationsimpl.CleanupServiceImpl)), cleanup.ProvideService, shorturlimpl.ProvideService, wire.Bind(new(shorturls.Service), new(*shorturlimpl.ShortURLService)), queryhistory.ProvideService, wire.Bind(new(queryhistory.Service), new(*queryhistory.QueryHistoryService)), correlations.ProvideService, wire.Bind(new(correlations.Service), new(*correlations.CorrelationsService)), quotaimpl.ProvideService, remotecache.ProvideService, wire.Bind(new(remotecache.CacheStorage), new(*remotecache.RemoteCache)), authinfoimpl.ProvideService, wire.Bind(new(login.AuthInfoService), new(*authinfoimpl.Service)), authinfoimpl.ProvideStore, datasourceproxy.ProvideService, sort.ProvideService, search2.ProvideService, searchV2.ProvideService, searchV2.ProvideSearchHTTPService, store.ProvideService, store.ProvideSystemUsersService, live.ProvideService, pushhttp.ProvideService, contexthandler.ProvideService, service12.ProvideService, wire.Bind(new(service12.LDAP), new(*service12.LDAPImpl)), jwt.ProvideService, wire.Bind(new(jwt.JWTService), new(*jwt.AuthService)), store2.ProvideDBStore, image.ProvideDeleteExpiredService, ngalert.ProvideService, librarypanels.ProvideService, wire.Bind(new(librarypanels.Service), new(*librarypanels.LibraryPanelService)), libraryelements.ProvideService, wire.Bind(new(libraryelements.Service), new(*libraryelements.LibraryElementService)), notifications.ProvideService, notifications.ProvideSmtpService, github.ProvideFactory, tracing.ProvideService, tracing.ProvideTracingConfig, wire.Bind(new(tracing.Tracer), new(*tracing.TracingService)), withOTelSet, testdatasource.ProvideService, api4.ProvideService, opentsdb.ProvideService, socialimpl.ProvideService, influxdb.ProvideService, wire.Bind(new(social.Service), new(*socialimpl.SocialService)), tempo.ProvideService, loki.ProvideService, graphite.ProvideService, prometheus.ProvideService, elasticsearch.ProvideService, pyroscope.ProvideService, parca.ProvideService, zipkin.ProvideService, jaeger.ProvideService, service9.ProvideCacheService, wire.Bind(new(datasources.CacheService), new(*service9.CacheServiceImpl)), service2.ProvideEncryptionService, wire.Bind(new(encryption2.Internal), new(*service2.Service)), manager.ProvideSecretsService, wire.Bind(new(secrets.Service), new(*manager.SecretsService)), database.ProvideSecretsStore, wire.Bind(new(secrets.Store), new(*database.SecretsStoreImpl)), garbagecollectionworker.ProvideWorker, grafanads.ProvideService, wire.Bind(new(dashboardsnapshots.Store), new(*database5.DashboardSnapshotStore)), database5.ProvideStore, wire.Bind(new(dashboardsnapshots.Service), new(*service10.ServiceImpl)), service10.ProvideService, service9.ProvideService, wire.Bind(new(datasources.DataSourceService), new(*service9.Service)), service9.ProvideLegacyDataSourceLookup, retriever.ProvideService, wire.Bind(new(serviceaccounts.ServiceAccountRetriever), new(*retriever.Service)), ossaccesscontrol.ProvideServiceAccountPermissions, wire.Bind(new(accesscontrol.ServiceAccountPermissionsService), new(*ossaccesscontrol.ServiceAccountPermissionsService)), manager3.ProvideServiceAccountsService, proxy.ProvideServiceAccountsProxy, wire.Bind(new(serviceaccounts.Service), new(*proxy.ServiceAccountsProxy)), dsquerierclient.NewNullQSDatasourceClientBuilder, expr.ProvideService, featuremgmt.ProvideManagerService, featuremgmt.ProvideToggles, service7.ProvideDashboardServiceImpl, wire.Bind(new(dashboards2.PermissionsRegistrationService), new(*service7.DashboardServiceImpl)), service7.ProvideDashboardService, service7.ProvideDashboardProvisioningService, service7.ProvideDashboardPluginService, database2.ProvideDashboardStore, folderimpl.ProvideService, wire.Bind(new(folder.Service), new(*folderimpl.Service)), wire.Bind(new(folder.LegacyService), new(*folderimpl.Service)), folderimpl.ProvideStore, wire.Bind(new(folder.Store), new(*folderimpl.FolderStoreImpl)), service11.ProvideService, wire.Bind(new(dashboardimport.Service), new(*service11.ImportDashboardService)), service8.ProvideService, wire.Bind(new(plugindashboards.Service), new(*service8.Service)), service8.ProvideDashboardUpdater, kvstore2.ProvideService, avatar.ProvideAvatarCacheServer, statscollector.ProvideService, csrf.ProvideCSRFFilter, wire.Bind(new(csrf.Service), new(*csrf.CSRF)), ossaccesscontrol.ProvideTeamPermissions, wire.Bind(new(accesscontrol.TeamPermissionsService), new(*ossaccesscontrol.TeamPermissionsService)), ossaccesscontrol.ProvideFolderPermissions, wire.Bind(new(accesscontrol.FolderPermissionsService), new(*ossaccesscontrol.FolderPermissionsService)), ossaccesscontrol.ProvideDashboardPermissions, wire.Bind(new(accesscontrol.DashboardPermissionsService), new(*ossaccesscontrol.DashboardPermissionsService)), ossaccesscontrol.ProvideReceiverPermissionsService, wire.Bind(new(accesscontrol.ReceiverPermissionsService), new(*ossaccesscontrol.ReceiverPermissionsService)), starimpl.ProvideService, playlistimpl.ProvideService, apikeyimpl.ProvideService, dashverimpl.ProvideService, service3.ProvideService, wire.Bind(new(publicdashboards.Service), new(*service3.PublicDashboardServiceImpl)), database3.ProvideStore, wire.Bind(new(publicdashboards.Store), new(*database3.PublicDashboardStoreImpl)), metric.ProvideService, api2.ProvideApi, api3.ProvideApi, userimpl.ProvideService, orgimpl.ProvideService, orgimpl.ProvideDeletionService, statsimpl.ProvideService, grpccontext.ProvideContextHandler, grpcserver.ProvideHealthService, grpcserver.ProvideReflectionService, resolver.ProvideEntityReferenceResolver, teamimpl.ProvideService, teamapi.ProvideTeamAPI, tempuserimpl.ProvideService, loginattemptimpl.ProvideService, wire.Bind(new(loginattempt.Service), new(*loginattemptimpl.Service)), migrations2.ProvideDataSourceMigrationService, migrations2.ProvideSecretMigrationProvider, wire.Bind(new(migrations2.SecretMigrationProvider), new(*migrations2.SecretMigrationProviderImpl)), promtypemigration.ProvideAzurePromMigrationService, promtypemigration.ProvideAmazonPromMigrationService, promtypemigration.ProvidePromTypeMigrationProvider, wire.Bind(new(promtypemigration.PromTypeMigrationProvider), new(*promtypemigration.PromTypeMigrationProviderImpl)), resourcepermissions.NewActionSetService, wire.Bind(new(accesscontrol.ActionResolver), new(resourcepermissions.ActionSetService)), wire.Bind(new(pluginaccesscontrol.ActionSetRegistry), new(resourcepermissions.ActionSetService)), permreg.ProvidePermissionRegistry, acimpl.ProvideAccessControl, accesscontrol.ProvideFixedRolesLoader, dualwrite2.ProvideZanzanaReconciler, navtreeimpl.ProvideService, wire.Bind(new(accesscontrol.AccessControl), new(*acimpl.AccessControl)), wire.Bind(new(notifications.TempUserStore), new(tempuser.Service)), tagimpl.ProvideService, wire.Bind(new(tag.Service), new(*tagimpl.Service)), authnimpl.ProvideService, authnimpl.ProvideIdentitySynchronizer, authnimpl.ProvideAuthnService, authnimpl.ProvideAuthnServiceAuthenticateOnly, authnimpl.ProvideRegistration, supportbundlesimpl.ProvideService, extsvcaccounts.ProvideExtSvcAccountsService, wire.Bind(new(serviceaccounts.ExtSvcAccountsService), new(*extsvcaccounts.ExtSvcAccountsService)), registry2.ProvideExtSvcRegistry, wire.Bind(new(extsvcauth.ExternalServiceRegistry), new(*registry2.Registry)), anonstore.ProvideAnonDBStore, wire.Bind(new(anonstore.AnonStore), new(*anonstore.AnonDBStore)), loggermw.Provide, slogadapter.Provide, signingkeysimpl.ProvideEmbeddedSigningKeysService, wire.Bind(new(signingkeys.Service), new(*signingkeysimpl.Service)), ssosettingsimpl.ProvideService, wire.Bind(new(ssosettings.Service), new(*ssosettingsimpl.Service)), idimpl.ProvideService, wire.Bind(new(auth.IDService), new(*idimpl.Service)), cloudmigrationimpl.ProvideService, caching.ProvideCachingServiceClient, userimpl.ProvideVerifier, connectors.ProvideOrgRoleMapper, wire.Bind(new(user.Verifier), new(*userimpl.Verifier)), authz.WireSet, metadata.ProvideSecureValueMetadataStorage, metadata.ProvideKeeperMetadataStorage, metadata.ProvideDecryptStorage, decrypt.ProvideDecryptAuthorizer, wire.Value([]decrypt.ExtraOwnerDecrypter(nil)), decrypt.ProvideDecryptService, inline.ProvideInlineSecureValueService, encryption.ProvideDataKeyStorage, encryption.ProvideGlobalDataKeyStorage, encryption.ProvideEncryptedValueStorage, encryption.ProvideGlobalEncryptedValueStorage, encryption.ProvideEncryptedValueMigrationExecutor, service5.ProvideSecureValueService, validator.ProvideKeeperValidator, validator.ProvideSecureValueValidator, mutator.ProvideKeeperMutator, mutator.ProvideSecureValueMutator, migrator.NewWithEngine, database4.ProvideDatabase, clock.ProvideClock, wire.Bind(new(contracts.Database), new(*database4.Database)), wire.Bind(new(contracts.Clock), new(*clock.Clock)), manager2.ProvideEncryptionManager, service4.ProvideAESGCMCipherService, resource.ProvideStorageMetrics, resource.ProvideIndexMetrics, migrations3.ProvideUnifiedStorageMigrationProvider, wire.Bind(new(migrations3.UnifiedStorageMigrationProvider), new(*migrations3.UnifiedStorageMigrationProviderImpl)), apiserver.WireSet, apiregistry.WireSet, appregistry.WireSet, client.ProvideK8sClientWithFallback) var wireSet = wire.NewSet( wireBasicSet, metrics.WireSet, sqlstore.ProvideService, metrics2.ProvideService, wire.Bind(new(notifications.Service), new(*notifications.NotificationService)), wire.Bind(new(notifications.WebhookSender), new(*notifications.NotificationService)), wire.Bind(new(notifications.EmailSender), new(*notifications.NotificationService)), wire.Bind(new(db.DB), new(*sqlstore.SQLStore)), prefimpl.ProvideService, oauthtoken.ProvideService, wire.Bind(new(oauthtoken.OAuthTokenService), new(*oauthtoken.Service)), wire.Bind(new(cleanup.AlertRuleService), new(*store2.DBstore)), diff --git a/pkg/storage/unified/client.go b/pkg/storage/unified/client.go index 0a4d3e630ff..07165937199 100644 --- a/pkg/storage/unified/client.go +++ b/pkg/storage/unified/client.go @@ -57,7 +57,6 @@ func ProvideUnifiedStorageClient(opts *Options, storageMetrics *resource.StorageMetrics, indexMetrics *resource.BleveIndexMetrics, ) (resource.ResourceClient, error) { - // See: apiserver.applyAPIServerConfig(cfg, features, o) apiserverCfg := opts.Cfg.SectionWithEnvOverrides("grafana-apiserver") client, err := newClient(options.StorageOptions{ StorageType: options.StorageType(apiserverCfg.Key("storage_type").MustString(string(options.StorageTypeUnified))), @@ -165,12 +164,8 @@ func newClient(opts options.StorageOptions, indexConn = conn } - // Create a client instance - client, err := resource.NewResourceClient(conn, indexConn, cfg, features, tracer) - if err != nil { - return nil, err - } - return client, nil + // Create a resource client + return resource.NewResourceClient(conn, indexConn, cfg, features, tracer) default: searchOptions, err := search.NewSearchOptions(features, cfg, tracer, docs, indexMetrics, nil) diff --git a/pkg/storage/unified/migrations/dashboard_folder_migration.go b/pkg/storage/unified/migrations/dashboard_folder_migration.go new file mode 100644 index 00000000000..96c5a824c8a --- /dev/null +++ b/pkg/storage/unified/migrations/dashboard_folder_migration.go @@ -0,0 +1,93 @@ +package migrations + +import ( + "context" + "fmt" + + "github.com/grafana/authlib/types" + "github.com/grafana/grafana/pkg/apimachinery/identity" + "github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy" + "github.com/grafana/grafana/pkg/services/sqlstore/migrator" + "github.com/grafana/grafana/pkg/storage/unified/resource" + "github.com/grafana/grafana/pkg/util/xorm" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +const ( + FoldersAndDashboardsMigrationID = "folders and dashboards migration" + UnifiedStorageDataMigrationSQL = "unified storage data migration" +) + +type dashboardAndFolderMigration struct { + migrator.MigrationBase + legacyMigrator legacy.LegacyMigrator + bulkStoreClient resource.ResourceClient +} + +var _ migrator.CodeMigration = (*dashboardAndFolderMigration)(nil) + +// SQL implements migrator.Migration interface. Returns a description string. +func (sp *dashboardAndFolderMigration) SQL(dialect migrator.Dialect) string { + return UnifiedStorageDataMigrationSQL +} + +func (sp *dashboardAndFolderMigration) Exec(sess *xorm.Session, mg *migrator.Migrator) error { + ctx := context.Background() + logger := mg.Logger + + resources := []schema.GroupResource{ + { + Group: "folder.grafana.app", + Resource: "folders", + }, + { + Group: "dashboard.grafana.app", + Resource: "dashboards", + }, + } + + storageMigrator := newUnifiedStorageMigrator(sp.legacyMigrator, sp.bulkStoreClient, resources, "unified-storage-migration.folders-dashboards") + + orgs, err := sp.getAllOrgs(sess) + if err != nil { + logger.Error("failed to get organizations for folders and dashboards migration", "error", err) + return fmt.Errorf("failed to get organizations: %w", err) + } + + if len(orgs) == 0 { + logger.Info("No organizations found to migrate, skipping migration") + return nil + } + + logger.Info("Starting migration for all organizations", "org_count", len(orgs)) + + for _, org := range orgs { + namespace := types.OrgNamespaceFormatter(org.ID) + logger.Info("Migrating organization", "org_id", org.ID, "org_name", org.Name, "namespace", namespace) + + // Create a service identity context for this namespace to authenticate with unified storage + migrationCtx, _ := identity.WithServiceIdentityForSingleNamespace(ctx, namespace) + + if err := storageMigrator.executeMigration(migrationCtx, sess, mg, namespace); err != nil { + logger.Error("migration failed for organization", "org_id", org.ID, "org_name", org.Name, "error", err) + return fmt.Errorf("migration failed for org %d (%s): %w", org.ID, org.Name, err) + } + } + + logger.Info("Migration completed successfully for all organizations", "org_count", len(orgs)) + return nil +} + +type orgInfo struct { + ID int64 `xorm:"id"` + Name string `xorm:"name"` +} + +func (sp *dashboardAndFolderMigration) getAllOrgs(sess *xorm.Session) ([]orgInfo, error) { + var orgs []orgInfo + err := sess.Table("org").Cols("id", "name").Find(&orgs) + if err != nil { + return nil, err + } + return orgs, nil +} diff --git a/pkg/storage/unified/migrations/migrations.go b/pkg/storage/unified/migrations/migrations.go new file mode 100644 index 00000000000..195b7f5d154 --- /dev/null +++ b/pkg/storage/unified/migrations/migrations.go @@ -0,0 +1,108 @@ +package migrations + +import ( + "context" + "fmt" + "os" + + "github.com/grafana/grafana/pkg/infra/db" + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/registry" + "github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy" + "github.com/grafana/grafana/pkg/services/sqlstore/migrator" + "github.com/grafana/grafana/pkg/setting" + "github.com/grafana/grafana/pkg/storage/unified/resource" + "github.com/prometheus/client_golang/prometheus" + "go.opentelemetry.io/otel" +) + +var tracer = otel.Tracer("github.com/grafana/grafana/pkg/storage/unified/migrations") + +// UnifiedStorageMigrationProvider provides unified storage migrations as a background service +type UnifiedStorageMigrationProvider interface { + registry.BackgroundService +} + +type UnifiedStorageMigrationProviderImpl struct { + legacyMigrator legacy.LegacyMigrator + cfg *setting.Cfg + client resource.ResourceClient + sqlStore db.DB +} + +var _ UnifiedStorageMigrationProvider = (*UnifiedStorageMigrationProviderImpl)(nil) + +// ProvideUnifiedStorageMigrationProvider is a Wire provider that creates the migration service. +// The service implements registry.BackgroundService and runs migrations during server startup. +func ProvideUnifiedStorageMigrationProvider( + legacyMigrator legacy.LegacyMigrator, + cfg *setting.Cfg, + client resource.ResourceClient, + sqlStore db.DB, +) *UnifiedStorageMigrationProviderImpl { + return &UnifiedStorageMigrationProviderImpl{ + legacyMigrator: legacyMigrator, + cfg: cfg, + client: client, + sqlStore: sqlStore, + } +} + +// Run executes unified storage migrations as a background service. +// This blocks until migrations complete. If migrations fail, an error is returned +// which will prevent Grafana from starting. +func (p *UnifiedStorageMigrationProviderImpl) Run(ctx context.Context) error { + // skip migrations in test environments to prevent integration test timeouts. + if os.Getenv("GRAFANA_TEST_DB") != "" { + return nil + } + + // TODO: Re-enable once migrations are ready + // return RegisterMigrations(p.legacyMigrator, p.cfg, p.client, p.sqlStore) + return nil +} + +// RegisterMigrations initializes and registers all unified storage migrations. +// This function is the entry point for all data migrations from legacy storage +// to unified storage. It returns an error if migrations fail, preventing Grafana +// from starting with inconsistent data. +func RegisterMigrations( + legacyMigrator legacy.LegacyMigrator, + cfg *setting.Cfg, + client resource.ResourceClient, + sqlStore db.DB, +) error { + ctx, span := tracer.Start(context.Background(), "storage.unified.RegisterMigrations") + defer span.End() + logger := log.New("storage.unified.migrations.folders-dashboards") + mg := migrator.NewScopedMigrator(sqlStore.GetEngine(), cfg, "unified_storage") + mg.AddCreateMigration() + + if err := prometheus.Register(mg); err != nil { + logger.Warn("Failed to register migrator metrics", "error", err) + } + + // Add new migration registrations here for each resource type + registerDashboardAndFolderMigration(mg, legacyMigrator, client) + + // Run all registered migrations (blocking) + sec := cfg.Raw.Section("database") + if err := mg.RunMigrations(ctx, 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) + } + + logger.Info("Unified storage migrations completed successfully") + return nil +} + +func registerDashboardAndFolderMigration( + mg *migrator.Migrator, + legacyMigrator legacy.LegacyMigrator, + bulkStoreClient resource.ResourceClient, +) { + migration := &dashboardAndFolderMigration{ + legacyMigrator: legacyMigrator, + bulkStoreClient: bulkStoreClient, + } + mg.AddMigration(FoldersAndDashboardsMigrationID, migration) +} diff --git a/pkg/storage/unified/migrations/migrator.go b/pkg/storage/unified/migrations/migrator.go new file mode 100644 index 00000000000..cab6ce38432 --- /dev/null +++ b/pkg/storage/unified/migrations/migrator.go @@ -0,0 +1,166 @@ +package migrations + +import ( + "context" + "fmt" + "time" + + "github.com/grafana/authlib/types" + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy" + "github.com/grafana/grafana/pkg/services/sqlstore/migrator" + "github.com/grafana/grafana/pkg/storage/unified/resource" + "github.com/grafana/grafana/pkg/storage/unified/resourcepb" + "github.com/grafana/grafana/pkg/util/xorm" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +// StorageMigrator defines the interface for executing unified storage migrations +type StorageMigrator interface { + executeMigration(ctx context.Context, sess *xorm.Session, mg *migrator.Migrator, namespace string) error +} + +type unifiedStorageMigrator struct { + migrator legacy.LegacyMigrator + bulkStoreClient resource.ResourceClient + resources []schema.GroupResource + log log.Logger +} + +func newUnifiedStorageMigrator(migrator legacy.LegacyMigrator, bulkStoreClient resource.ResourceClient, resources []schema.GroupResource, logPrefix string) StorageMigrator { + return &unifiedStorageMigrator{ + migrator: migrator, + bulkStoreClient: bulkStoreClient, + resources: resources, + log: log.New(logPrefix), + } +} + +func (m *unifiedStorageMigrator) executeMigration(ctx context.Context, sess *xorm.Session, mg *migrator.Migrator, namespace string) error { + startTime := time.Now() + m.log.Info("Starting unified storage migration", "namespace", namespace, "resources", m.resources) + + opts := legacy.MigrateOptions{ + Namespace: namespace, + Store: m.bulkStoreClient, + LargeObjects: nil, // Not using large object support to avoid import cycles + Resources: m.resources, + WithHistory: true, // Migrate with full history + OnlyCount: false, + Progress: func(count int, msg string) { + m.log.Info("Migration progress", "count", count, "message", msg) + }, + } + + // Execute the migration via legacy migrator + response, err := m.migrator.Migrate(ctx, opts) + if err != nil { + m.log.Error("Migration failed", "error", err, "duration", time.Since(startTime)) + return fmt.Errorf("failed to migrate resources: %w", err) + } + + // Validate the migration results + if err := m.validateMigration(sess, response); err != nil { + m.log.Error("Migration validation failed", "error", err, "duration", time.Since(startTime)) + return fmt.Errorf("migration validation failed: %w", err) + } + + m.log.Info("Migration completed successfully", + "duration", time.Since(startTime), + "processed", response.Processed, + "summaries", len(response.Summary), + "rejected", len(response.Rejected)) + + return nil +} + +func (m *unifiedStorageMigrator) validateMigration(sess *xorm.Session, response *resourcepb.BulkResponse) error { + // Check for rejected items + if len(response.Rejected) > 0 { + m.log.Warn("Migration had rejected items", "count", len(response.Rejected)) + for i, rejected := range response.Rejected { + if i < 10 { // Log first 10 rejected items + m.log.Warn("Rejected item", + "namespace", rejected.Key.Namespace, + "group", rejected.Key.Group, + "resource", rejected.Key.Resource, + "name", rejected.Key.Name, + "reason", rejected.Error) + } + } + // Rejections are not fatal - they may be expected for invalid data + } + + // Validate counts for each resource type + for _, summary := range response.Summary { + legacyCount, err := m.getLegacyCount(sess, summary.Group, summary.Resource, summary.Namespace) + if err != nil { + return fmt.Errorf("failed to get legacy count for %s/%s: %w", summary.Group, summary.Resource, err) + } + + // Account for rejected items in validation + expectedCount := summary.Count + int64(len(response.Rejected)) + + m.log.Info("Count validation", + "resource", fmt.Sprintf("%s.%s", summary.Resource, summary.Group), + "namespace", summary.Namespace, + "legacy_count", legacyCount, + "unified_count", summary.Count, + "rejected", len(response.Rejected), + "history", summary.History) + + // Validate that we migrated all items (allowing for rejected items) + if legacyCount > expectedCount { + return fmt.Errorf("count mismatch for %s.%s in namespace %s: legacy has %d, unified has %d, rejected %d", + summary.Resource, summary.Group, summary.Namespace, + legacyCount, summary.Count, len(response.Rejected)) + } + } + + return nil +} + +func (m *unifiedStorageMigrator) getLegacyCount(sess *xorm.Session, group, resourceType, namespace string) (int64, error) { + // Parse namespace to get org ID + orgID, err := ParseOrgIDFromNamespace(namespace) + if err != nil { + return 0, fmt.Errorf("invalid namespace %s: %w", namespace, err) + } + + // Map group/resource to legacy table + tableName, whereClause := m.getLegacyTableInfo(group, resourceType) + if tableName == "" { + return 0, fmt.Errorf("unknown resource type: %s.%s", resourceType, group) + } + + // Count items in legacy table using Table() before Count() + count, err := sess.Table(tableName).Where(whereClause, orgID).Count() + if err != nil { + return 0, fmt.Errorf("failed to count %s: %w", tableName, err) + } + + return count, nil +} + +func (m *unifiedStorageMigrator) getLegacyTableInfo(group, resource string) (table string, whereClause string) { + // Map unified storage group/resource to legacy tables + switch { + case group == "dashboard.grafana.app" && resource == "dashboards": + return "dashboard", "org_id = ? and is_folder = false" + case group == "folder.grafana.app" && resource == "folders": + return "dashboard", "org_id = ? and is_folder = true" + case group == "playlist.grafana.app" && resource == "playlists": + return "playlist", "org_id = ?" + default: + return "", "" + } +} + +func ParseOrgIDFromNamespace(namespace string) (int64, error) { + // Use authlib to properly parse all namespace formats including "default" for org 1 + info, err := types.ParseNamespace(namespace) + if err != nil { + return 0, fmt.Errorf("failed to parse namespace: %w", err) + } + return info.OrgID, nil +}