public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add unified data migrations for dashboard and folders (#113853)

Browse Source
This commit is contained in:
Mustafa Sencer Özcan
2025-11-19 09:09:08 +01:00
committed by GitHub
parent 92bee5e510
commit 0d4ad01b65
8 changed files with 385 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/dashboard/legacy/migrate.go
+2 -5
View File
@@ -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))
}
pkg/registry/backgroundsvcs/background_services.go
+3
View File
@@ -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,
pkg/server/wire.go
+3
View File
@@ -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,
pkg/server/wire_gen.go
Generated
+8 -5
View File
File diff suppressed because one or more lines are too long
pkg/storage/unified/client.go
+2 -7
View File
@@ -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)
pkg/storage/unified/migrations/dashboard_folder_migration.go
+93
View File
@@ -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
}
pkg/storage/unified/migrations/migrations.go
+108
View File
@@ -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)
}
pkg/storage/unified/migrations/migrator.go
+166
View File
@@ -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
}