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

fix(unified): in-proc SQLite data migration (#114537)

Browse Source 
* feat: unified storage migrations integration tests

* chore: add comment and adjust db path name

* chore: refactor test cases into interface

* fix: unified SQLite migration with SQLStore migrator

* revert changes to newResourceDBProvider
This commit is contained in:
Rafael Bortolon Paulovic
2025-11-28 13:13:35 +01:00
committed by GitHub
parent 11a27ab870
commit 12c6d7e83f
7 changed files with 286 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/storage/unified/migrations/resource_migration.go
+16
View File
@@ -10,6 +10,7 @@ import (
"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"
@@ -72,6 +73,17 @@ func (m *ResourceMigration) Exec(sess *xorm.Session, mg *migrator.Migrator) erro
m.log.Info("Starting migration for all organizations", "org_count", len(orgs), "resources", m.resources)
if mg.Dialect.DriverName() == migrator.SQLite {
// reuse transaction in SQLite to avoid "database is locked" errors
tx, err := sess.Tx()
if err != nil {
m.log.Error("Failed to get transaction from session", "error", err)
return fmt.Errorf("failed to get transaction: %w", err)
}
ctx = resource.ContextWithTransaction(ctx, tx.Tx)
m.log.Info("Stored migrator transaction in context for bulk operations (SQLite compatibility)")
}
for _, org := range orgs {
if err := m.migrateOrg(ctx, sess, org); err != nil {
return err
@@ -107,6 +119,10 @@ func (m *ResourceMigration) migrateOrg(ctx context.Context, sess *xorm.Session,
m.log.Error("Migration failed", "org_id", org.ID, "error", err, "duration", time.Since(startTime))
return fmt.Errorf("migration failed for org %d (%s): %w", org.ID, org.Name, err)
}
if response.Error != nil {
m.log.Error("Migration reported error", "org_id", org.ID, "error", response.Error.String(), "duration", time.Since(startTime))
return fmt.Errorf("migration failed for org %d (%s): %w", org.ID, org.Name, fmt.Errorf("migration error: %s", response.Error.Message))
}
// Validate the migration results
if err := m.validateMigration(migrationCtx, sess, response); err != nil {
pkg/storage/unified/migrations/service.go
+10 -2
View File
@@ -85,8 +85,13 @@ func RegisterMigrations(
// 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,
sec.Key("migration_locking").MustBool(true),
migrationLocking,
sec.Key("locking_attempt_timeout_sec").MustInt()); err != nil {
return fmt.Errorf("unified storage data migration failed: %w", err)
}
@@ -98,12 +103,14 @@ 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"}
driverName := mg.Dialect.DriverName()
folderCountValidator := NewCountValidator(
client,
folders,
"dashboard",
"org_id = ? and is_folder = true",
driverName,
)
dashboardCountValidator := NewCountValidator(
@@ -111,9 +118,10 @@ func registerDashboardAndFolderMigration(mg *sqlstoremigrator.Migrator, migrator
dashboards,
"dashboard",
"org_id = ? and is_folder = false",
driverName,
)
folderTreeValidator := NewFolderTreeValidator(client, folders)
folderTreeValidator := NewFolderTreeValidator(client, folders, driverName)
dashboardsAndFolders := NewResourceMigration(
migrator,
pkg/storage/unified/migrations/validator.go
+77 -22
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/services/sqlstore/migrator"
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
"github.com/grafana/grafana/pkg/util/xorm"
"k8s.io/apimachinery/pkg/runtime/schema"
@@ -57,6 +58,7 @@ type CountValidator struct {
resource schema.GroupResource
table string
whereClause string
driverName string
}
func NewCountValidator(
@@ -64,6 +66,7 @@ func NewCountValidator(
resource schema.GroupResource,
table string,
whereClause string,
driverName string,
) Validator {
return &CountValidator{
name: "CountValidator",
@@ -71,6 +74,7 @@ func NewCountValidator(
resource: resource,
table: table,
whereClause: whereClause,
driverName: driverName,
}
}
@@ -120,22 +124,32 @@ func (v *CountValidator) Validate(ctx context.Context, sess *xorm.Session, respo
return fmt.Errorf("failed to count %s: %w", v.table, err)
}
// Get unified storage count using GetStats API
statsResp, err := v.client.GetStats(ctx, &resourcepb.ResourceStatsRequest{
Namespace: summary.Namespace,
Kinds: []string{fmt.Sprintf("%s/%s", summary.Group, summary.Resource)},
})
if err != nil {
return fmt.Errorf("failed to get stats for %s/%s in namespace %s: %w",
summary.Group, summary.Resource, summary.Namespace, err)
}
// Find the count for this specific resource type
var unifiedCount int64
for _, stat := range statsResp.Stats {
if stat.Group == summary.Group && stat.Resource == summary.Resource {
unifiedCount = stat.Count
break
if v.driverName == migrator.SQLite {
unifiedCount, err = sess.Table("resource").
Where("namespace = ? AND `group` = ? AND resource = ?",
summary.Namespace, summary.Group, summary.Resource).
Count()
if err != nil {
return fmt.Errorf("failed to count resource table for %s/%s in namespace %s: %w",
summary.Group, summary.Resource, summary.Namespace, err)
}
} else {
// Get unified storage count using GetStats API
statsResp, err := v.client.GetStats(ctx, &resourcepb.ResourceStatsRequest{
Namespace: summary.Namespace,
Kinds: []string{fmt.Sprintf("%s/%s", summary.Group, summary.Resource)},
})
if err != nil {
return fmt.Errorf("failed to get stats for %s/%s in namespace %s: %w",
summary.Group, summary.Resource, summary.Namespace, err)
}
// Find the count for this specific resource type
for _, stat := range statsResp.Stats {
if stat.Group == summary.Group && stat.Resource == summary.Resource {
unifiedCount = stat.Count
break
}
}
}
@@ -162,19 +176,22 @@ func (v *CountValidator) Validate(ctx context.Context, sess *xorm.Session, respo
}
type FolderTreeValidator struct {
name string
client resourcepb.ResourceIndexClient
resource schema.GroupResource
name string
client resourcepb.ResourceIndexClient
resource schema.GroupResource
driverName string
}
func NewFolderTreeValidator(
client resourcepb.ResourceIndexClient,
resource schema.GroupResource,
driverName string,
) Validator {
return &FolderTreeValidator{
name: "FolderTreeValidator",
client: client,
resource: resource,
name: "FolderTreeValidator",
client: client,
resource: resource,
driverName: driverName,
}
}
@@ -185,6 +202,12 @@ type legacyFolder struct {
Title string `xorm:"title"`
}
type unifiedFolder struct {
GUID string `xorm:"guid"`
Name string `xorm:"name"`
Folder string `xorm:"folder"`
}
func (v *FolderTreeValidator) Name() string {
return v.name
}
@@ -218,7 +241,12 @@ func (v *FolderTreeValidator) Validate(ctx context.Context, sess *xorm.Session,
}
// Build unified storage folder parent map
unifiedParentMap, err := v.buildUnifiedFolderParentMap(ctx, summary.Namespace, log)
var unifiedParentMap map[string]string
if v.driverName == migrator.SQLite {
unifiedParentMap, err = v.buildUnifiedFolderParentMapSQLite(sess, summary.Namespace, log)
} else {
unifiedParentMap, err = v.buildUnifiedFolderParentMap(ctx, summary.Namespace, log)
}
if err != nil {
return fmt.Errorf("failed to build unified folder parent map: %w", err)
}
@@ -348,3 +376,30 @@ func (v *FolderTreeValidator) buildUnifiedFolderParentMap(ctx context.Context, n
return parentMap, nil
}
func (v *FolderTreeValidator) buildUnifiedFolderParentMapSQLite(sess *xorm.Session, namespace string, log log.Logger) (map[string]string, error) {
var folders []unifiedFolder
err := sess.Table("resource").
Cols("guid", "name", "folder").
Where("namespace = ? AND resource = ?", namespace, "folder").
Find(&folders)
if err != nil {
return nil, fmt.Errorf("failed to query unified folders: %w", err)
}
parentMap := make(map[string]string)
for _, folder := range folders {
parentMap[folder.Name] = folder.Folder
}
if len(parentMap) == 0 {
log.Debug("No unified folders found for namespace", "namespace", namespace)
return make(map[string]string), nil
}
log.Debug("Built unified folder parent map",
"folder_count", len(parentMap),
"namespace", namespace)
return parentMap, nil
}
pkg/storage/unified/resource/transaction.go
+25
View File
@@ -0,0 +1,25 @@
package resource
import (
"context"
"database/sql"
)
type transactionContextKey struct{}
// ContextWithTransaction returns a new context with the transaction stored directly.
// This is used for SQLite migrations where the transaction needs to be shared
// between the migration code and unified storage operations within the same process.
func ContextWithTransaction(ctx context.Context, tx *sql.Tx) context.Context {
return context.WithValue(ctx, transactionContextKey{}, tx)
}
// TransactionFromContext retrieves the transaction from context
func TransactionFromContext(ctx context.Context) *sql.Tx {
if v := ctx.Value(transactionContextKey{}); v != nil {
if tx, ok := v.(*sql.Tx); ok {
return tx
}
}
return nil
}
pkg/storage/unified/sql/bulk.go
+143 -107
View File
@@ -8,6 +8,7 @@ import (
"sync"
"time"
"github.com/fullstorydev/grpchan/inprocgrpc"
"github.com/google/uuid"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -20,6 +21,7 @@ import (
"github.com/grafana/grafana/pkg/storage/unified/resource"
"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
"github.com/grafana/grafana/pkg/storage/unified/sql/db"
"github.com/grafana/grafana/pkg/storage/unified/sql/db/dbimpl"
"github.com/grafana/grafana/pkg/storage/unified/sql/dbutil"
"github.com/grafana/grafana/pkg/storage/unified/sql/sqltemplate"
)
@@ -111,6 +113,19 @@ func (b *backend) ProcessBulk(ctx context.Context, setting resource.BulkSettings
}
defer b.bulkLock.Finish(setting.Collection)
// If provided, reuse the inproc transaction for SQLite
if clientCtx := inprocgrpc.ClientContext(ctx); clientCtx != nil && b.dialect.DialectName() == "sqlite" {
if externalTx := resource.TransactionFromContext(clientCtx); externalTx != nil {
b.log.Info("Using SQLite transaction from client context")
rsp := &resourcepb.BulkResponse{}
err := b.processBulkWithTx(ctx, dbimpl.NewTx(externalTx), setting, iter, rsp)
if err != nil {
rsp.Error = resource.AsErrorResult(err)
}
return rsp
}
}
// We may want to first write parquet, then read parquet
if b.dialect.DialectName() == "sqlite" {
file, err := os.CreateTemp("", "grafana-bulk-export-*.parquet")
@@ -151,109 +166,134 @@ func (b *backend) ProcessBulk(ctx context.Context, setting resource.BulkSettings
func (b *backend) processBulk(ctx context.Context, setting resource.BulkSettings, iter resource.BulkRequestIterator) *resourcepb.BulkResponse {
rsp := &resourcepb.BulkResponse{}
err := b.db.WithTx(ctx, ReadCommitted, func(ctx context.Context, tx db.Tx) error {
rollbackWithError := func(err error) error {
txerr := tx.Rollback()
if txerr != nil {
b.log.Warn("rollback", "error", txerr)
} else {
b.log.Info("rollback")
return b.processBulkWithTx(ctx, tx, setting, iter, rsp)
})
if err != nil {
rsp.Error = resource.AsErrorResult(err)
}
return rsp
}
// processBulkWithTx performs the bulk operation using the provided transaction.
// This is used both when creating our own transaction and when reusing an external one.
func (b *backend) processBulkWithTx(ctx context.Context, tx db.Tx, setting resource.BulkSettings, iter resource.BulkRequestIterator, rsp *resourcepb.BulkResponse) error {
rollbackWithError := func(err error) error {
txerr := tx.Rollback()
if txerr != nil {
b.log.Warn("rollback", "error", txerr)
} else {
b.log.Info("rollback")
}
return err
}
bulk := &bulkWroker{
ctx: ctx,
tx: tx,
dialect: b.dialect,
logger: logging.FromContext(ctx),
}
// Calculate the RV based on incoming request timestamps
rv := newBulkRV()
summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection))
// First clear everything in the transaction
if setting.RebuildCollection {
for _, key := range setting.Collection {
summary, err := bulk.deleteCollection(key)
if err != nil {
return rollbackWithError(err)
}
summaries[resource.NSGR(key)] = summary
rsp.Summary = append(rsp.Summary, summary)
}
} else {
for _, key := range setting.Collection {
summaries[resource.NSGR(key)] = &resourcepb.BulkResponse_Summary{
Namespace: key.Namespace,
Group: key.Group,
Resource: key.Resource,
}
}
}
obj := &unstructured.Unstructured{}
// Write each event into the history
for iter.Next() {
if iter.RollbackRequested() {
return rollbackWithError(nil)
}
req := iter.Request()
if req == nil {
return rollbackWithError(fmt.Errorf("missing request"))
}
rsp.Processed++
if req.Action == resourcepb.BulkRequest_UNKNOWN {
rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{
Key: req.Key,
Action: req.Action,
Error: "unknown action",
})
continue
}
err := obj.UnmarshalJSON(req.Value)
if err != nil {
rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{
Key: req.Key,
Action: req.Action,
Error: "unable to unmarshal json",
})
continue
}
// Write the event to history
if _, err := dbutil.Exec(ctx, tx, sqlResourceHistoryInsert, sqlResourceRequest{
SQLTemplate: sqltemplate.New(b.dialect),
WriteEvent: resource.WriteEvent{
Key: req.Key,
Type: resourcepb.WatchEvent_Type(req.Action),
Value: req.Value,
PreviousRV: -1, // Used for WATCH, but we want to skip watch events
},
Folder: req.Folder,
GUID: uuid.New().String(),
ResourceVersion: rv.next(obj),
}); err != nil {
return rollbackWithError(fmt.Errorf("insert into resource history: %w", err))
}
}
// Now update the resource table from history
for _, key := range setting.Collection {
k := fmt.Sprintf("%s/%s/%s", key.Namespace, key.Group, key.Resource)
summary := summaries[k]
if summary == nil {
return rollbackWithError(fmt.Errorf("missing summary key for: %s", k))
}
err := bulk.syncCollection(key, summary)
if err != nil {
return err
}
bulk := &bulkWroker{
ctx: ctx,
tx: tx,
dialect: b.dialect,
logger: logging.FromContext(ctx),
}
// Calculate the RV based on incoming request timestamps
rv := newBulkRV()
summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection))
// First clear everything in the transaction
if setting.RebuildCollection {
for _, key := range setting.Collection {
summary, err := bulk.deleteCollection(key)
if err != nil {
return rollbackWithError(err)
if b.dialect.DialectName() == "sqlite" {
nextRV, err := b.rvManager.lock(ctx, tx, key.Group, key.Resource)
if err != nil {
b.log.Error("error locking RV", "error", err, "key", resource.NSGR(key))
} else {
b.log.Info("successfully locked RV", "nextRV", nextRV, "key", resource.NSGR(key))
// Save the incremented RV
if err := b.rvManager.saveRV(ctx, tx, key.Group, key.Resource, nextRV); err != nil {
b.log.Error("error saving RV", "error", err, "key", resource.NSGR(key))
} else {
b.log.Info("successfully saved RV", "rv", nextRV, "key", resource.NSGR(key))
}
summaries[resource.NSGR(key)] = summary
rsp.Summary = append(rsp.Summary, summary)
}
} else {
for _, key := range setting.Collection {
summaries[resource.NSGR(key)] = &resourcepb.BulkResponse_Summary{
Namespace: key.Namespace,
Group: key.Group,
Resource: key.Resource,
}
}
}
obj := &unstructured.Unstructured{}
// Write each event into the history
for iter.Next() {
if iter.RollbackRequested() {
return rollbackWithError(nil)
}
req := iter.Request()
if req == nil {
return rollbackWithError(fmt.Errorf("missing request"))
}
rsp.Processed++
if req.Action == resourcepb.BulkRequest_UNKNOWN {
rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{
Key: req.Key,
Action: req.Action,
Error: "unknown action",
})
continue
}
err := obj.UnmarshalJSON(req.Value)
if err != nil {
rsp.Rejected = append(rsp.Rejected, &resourcepb.BulkResponse_Rejected{
Key: req.Key,
Action: req.Action,
Error: "unable to unmarshal json",
})
continue
}
// Write the event to history
if _, err := dbutil.Exec(ctx, tx, sqlResourceHistoryInsert, sqlResourceRequest{
SQLTemplate: sqltemplate.New(b.dialect),
WriteEvent: resource.WriteEvent{
Key: req.Key,
Type: resourcepb.WatchEvent_Type(req.Action),
Value: req.Value,
PreviousRV: -1, // Used for WATCH, but we want to skip watch events
},
Folder: req.Folder,
GUID: uuid.New().String(),
ResourceVersion: rv.next(obj),
}); err != nil {
return rollbackWithError(fmt.Errorf("insert into resource history: %w", err))
}
}
// Now update the resource table from history
for _, key := range setting.Collection {
k := fmt.Sprintf("%s/%s/%s", key.Namespace, key.Group, key.Resource)
summary := summaries[k]
if summary == nil {
return rollbackWithError(fmt.Errorf("missing summary key for: %s", k))
}
err := bulk.syncCollection(key, summary)
if err != nil {
return err
}
// Make sure the collection RV is above our last written event
_, err = b.rvManager.ExecWithRV(ctx, key, func(tx db.Tx) (string, error) {
return "", nil
@@ -261,19 +301,15 @@ func (b *backend) processBulk(ctx context.Context, setting resource.BulkSettings
if err != nil {
b.log.Warn("error increasing RV", "error", err)
}
// Update the last import time. This is important to trigger reindexing
// of the resource for a given namespace.
if err := b.updateLastImportTime(ctx, tx, key, time.Now()); err != nil {
return rollbackWithError(err)
}
}
return nil
})
if err != nil {
rsp.Error = resource.AsErrorResult(err)
// Update the last import time. This is important to trigger reindexing
// of the resource for a given namespace.
if err := b.updateLastImportTime(ctx, tx, key, time.Now()); err != nil {
return rollbackWithError(err)
}
}
return rsp
return nil
}
func (b *backend) updateLastImportTime(ctx context.Context, tx db.Tx, key *resourcepb.ResourceKey, now time.Time) error {
pkg/storage/unified/sql/db/dbimpl/db.go
+5
View File
@@ -48,6 +48,11 @@ type sqlTx struct {
*sql.Tx
}
// NewTx wraps an existing *sql.Tx with sqlTx
func NewTx(tx *sql.Tx) db.Tx {
return sqlTx{tx}
}
func (tx sqlTx) QueryContext(ctx context.Context, query string, args ...any) (db.Rows, error) {
// // codeql-suppress go/sql-query-built-from-user-controlled-sources "The query comes from a safe template source
// and the parameters are passed as arguments."
pkg/util/xorm/session.go
+10 -1
View File
@@ -7,6 +7,7 @@ package xorm
import (
"context"
"database/sql"
"errors"
"fmt"
"hash/crc32"
"reflect"
@@ -43,7 +44,7 @@ type Session struct {
afterProcessors []executedProcessor
prepareStmt bool
stmtCache map[uint32]*core.Stmt //key: hash.Hash32 of (queryStr, len(queryStr))
stmtCache map[uint32]*core.Stmt // key: hash.Hash32 of (queryStr, len(queryStr))
// !evalphobia! stored the last executed query on this session
lastSQL string
@@ -236,6 +237,14 @@ func (session *Session) DB() *core.DB {
return session.db
}
// Tx returns the underlying transaction
func (session *Session) Tx() (*core.Tx, error) {
if session.tx == nil {
return nil, errors.New("no open transaction")
}
return session.tx, nil
}
func cleanupProcessorsClosures(slices *[]func(any)) {
if len(*slices) > 0 {
*slices = make([]func(any), 0)