diff --git a/packages/grafana-data/src/types/featureToggles.gen.ts b/packages/grafana-data/src/types/featureToggles.gen.ts index 8e955663673..a7dd0dfe04f 100644 --- a/packages/grafana-data/src/types/featureToggles.gen.ts +++ b/packages/grafana-data/src/types/featureToggles.gen.ts @@ -214,6 +214,7 @@ export interface FeatureToggles { unifiedStorageSearch?: boolean; unifiedStorageSearchSprinkles?: boolean; unifiedStorageSearchPermissionFiltering?: boolean; + managedDualWriter?: boolean; pluginsSriChecks?: boolean; unifiedStorageBigObjectsSupport?: boolean; timeRangeProvider?: boolean; diff --git a/pkg/server/wire.go b/pkg/server/wire.go index 2dd9caa927c..d7ea2b2b577 100644 --- a/pkg/server/wire.go +++ b/pkg/server/wire.go @@ -157,6 +157,7 @@ import ( "github.com/grafana/grafana/pkg/services/user" "github.com/grafana/grafana/pkg/services/user/userimpl" "github.com/grafana/grafana/pkg/setting" + legacydualwrite "github.com/grafana/grafana/pkg/storage/legacysql/dualwrite" "github.com/grafana/grafana/pkg/storage/unified" "github.com/grafana/grafana/pkg/storage/unified/resource" unifiedsearch "github.com/grafana/grafana/pkg/storage/unified/search" @@ -215,6 +216,7 @@ var wireBasicSet = wire.NewSet( mssql.ProvideService, store.ProvideEntityEventsService, unified.ProvideUnifiedStorageClient, + legacydualwrite.ProvideService, httpclientprovider.New, wire.Bind(new(httpclient.Provider), new(*sdkhttpclient.Provider)), serverlock.ProvideService, diff --git a/pkg/services/apiserver/builder/helper.go b/pkg/services/apiserver/builder/helper.go index 5e76c41fac5..94f267c5544 100644 --- a/pkg/services/apiserver/builder/helper.go +++ b/pkg/services/apiserver/builder/helper.go @@ -33,6 +33,7 @@ import ( grafanarest "github.com/grafana/grafana/pkg/apiserver/rest" "github.com/grafana/grafana/pkg/services/apiserver/endpoints/request" "github.com/grafana/grafana/pkg/services/apiserver/options" + "github.com/grafana/grafana/pkg/storage/legacysql/dualwrite" "github.com/grafana/grafana/pkg/storage/unified/apistore" ) @@ -276,6 +277,7 @@ func InstallAPIs( namespaceMapper request.NamespaceMapper, kvStore grafanarest.NamespacedKVStore, serverLock ServerLockService, + dualWriteService dualwrite.Service, optsregister apistore.StorageOptionsRegister, ) error { // dual writing is only enabled when the storage type is not legacy. @@ -286,6 +288,11 @@ func InstallAPIs( // nolint:staticcheck if storageOpts.StorageType != options.StorageTypeLegacy { dualWrite = func(gr schema.GroupResource, legacy grafanarest.LegacyStorage, storage grafanarest.Storage) (grafanarest.Storage, error) { + // Dashboards + Folders may be managed (depends on feature toggles and database state) + if dualWriteService.ShouldManage(gr) { + return dualWriteService.NewStorage(gr, legacy, storage) // eventually this can replace this whole function + } + key := gr.String() // ${resource}.{group} eg playlists.playlist.grafana.app // Get the option from custom.ini/command line diff --git a/pkg/services/apiserver/service.go b/pkg/services/apiserver/service.go index 9246d08439a..0c09ad4afa4 100644 --- a/pkg/services/apiserver/service.go +++ b/pkg/services/apiserver/service.go @@ -47,6 +47,7 @@ import ( "github.com/grafana/grafana/pkg/services/org" "github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore" "github.com/grafana/grafana/pkg/setting" + "github.com/grafana/grafana/pkg/storage/legacysql/dualwrite" "github.com/grafana/grafana/pkg/storage/unified/apistore" "github.com/grafana/grafana/pkg/storage/unified/resource" ) @@ -137,6 +138,7 @@ type service struct { authorizer *authorizer.GrafanaAuthorizer serverLockService builder.ServerLockService + storageStatus dualwrite.Service kvStore kvstore.KVStore pluginClient plugins.Client @@ -159,6 +161,7 @@ func ProvideService( datasources datasource.ScopedPluginDatasourceProvider, contextProvider datasource.PluginContextWrapper, pluginStore pluginstore.Store, + storageStatus dualwrite.Service, unified resource.ResourceClient, ) (*service, error) { s := &service{ @@ -178,6 +181,7 @@ func ProvideService( contextProvider: contextProvider, pluginStore: pluginStore, serverLockService: serverLockService, + storageStatus: storageStatus, unified: unified, } // This will be used when running as a dskit service @@ -361,6 +365,7 @@ func (s *service) start(ctx context.Context) error { // Required for the dual writer initialization s.metrics, request.GetNamespaceMapper(s.cfg), kvstore.WithNamespace(s.kvStore, 0, "storage.dualwriting"), s.serverLockService, + s.storageStatus, optsregister, ) if err != nil { diff --git a/pkg/services/featuremgmt/registry.go b/pkg/services/featuremgmt/registry.go index 879c3153378..8f1d3370e1c 100644 --- a/pkg/services/featuremgmt/registry.go +++ b/pkg/services/featuremgmt/registry.go @@ -1483,6 +1483,14 @@ var ( HideFromDocs: true, HideFromAdminPage: true, }, + { + Name: "managedDualWriter", + Description: "Pick the dual write mode from database configs", + Stage: FeatureStageExperimental, + Owner: grafanaSearchAndStorageSquad, + HideFromDocs: true, + HideFromAdminPage: true, + }, { Name: "pluginsSriChecks", Description: "Enables SRI checks for plugin assets", diff --git a/pkg/services/featuremgmt/toggles_gen.csv b/pkg/services/featuremgmt/toggles_gen.csv index a3694fa49ae..f91670e92e4 100644 --- a/pkg/services/featuremgmt/toggles_gen.csv +++ b/pkg/services/featuremgmt/toggles_gen.csv @@ -195,6 +195,7 @@ rolePickerDrawer,experimental,@grafana/identity-access-team,false,false,false unifiedStorageSearch,experimental,@grafana/search-and-storage,false,false,false unifiedStorageSearchSprinkles,experimental,@grafana/search-and-storage,false,false,false unifiedStorageSearchPermissionFiltering,experimental,@grafana/search-and-storage,false,false,false +managedDualWriter,experimental,@grafana/search-and-storage,false,false,false pluginsSriChecks,experimental,@grafana/plugins-platform-backend,false,false,false unifiedStorageBigObjectsSupport,experimental,@grafana/search-and-storage,false,false,false timeRangeProvider,experimental,@grafana/grafana-frontend-platform,false,false,false diff --git a/pkg/services/featuremgmt/toggles_gen.go b/pkg/services/featuremgmt/toggles_gen.go index d2f0868a839..d144190409e 100644 --- a/pkg/services/featuremgmt/toggles_gen.go +++ b/pkg/services/featuremgmt/toggles_gen.go @@ -791,6 +791,10 @@ const ( // Enable permission filtering on unified storage search FlagUnifiedStorageSearchPermissionFiltering = "unifiedStorageSearchPermissionFiltering" + // FlagManagedDualWriter + // Pick the dual write mode from database configs + FlagManagedDualWriter = "managedDualWriter" + // FlagPluginsSriChecks // Enables SRI checks for plugin assets FlagPluginsSriChecks = "pluginsSriChecks" diff --git a/pkg/services/featuremgmt/toggles_gen.json b/pkg/services/featuremgmt/toggles_gen.json index 0cdcc21124f..d082c839f38 100644 --- a/pkg/services/featuremgmt/toggles_gen.json +++ b/pkg/services/featuremgmt/toggles_gen.json @@ -2611,6 +2611,20 @@ "expression": "true" } }, + { + "metadata": { + "name": "managedDualWriter", + "resourceVersion": "1739347092893", + "creationTimestamp": "2025-02-12T07:58:12Z" + }, + "spec": { + "description": "Pick the dual write mode from database configs", + "stage": "experimental", + "codeowner": "@grafana/search-and-storage", + "hideFromAdminPage": true, + "hideFromDocs": true + } + }, { "metadata": { "name": "managedPluginsInstall", diff --git a/pkg/storage/legacysql/dualwrite/managed_mode3.go b/pkg/storage/legacysql/dualwrite/managed_mode3.go new file mode 100644 index 00000000000..af1cee40613 --- /dev/null +++ b/pkg/storage/legacysql/dualwrite/managed_mode3.go @@ -0,0 +1,134 @@ +package dualwrite + +import ( + "context" + "net/http" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apiserver/pkg/registry/rest" + + grafanarest "github.com/grafana/grafana/pkg/apiserver/rest" +) + +func (m *service) NewStorage(gr schema.GroupResource, + legacy grafanarest.LegacyStorage, + storage grafanarest.Storage, +) (grafanarest.Storage, error) { + status, _ := m.Status(context.Background(), gr) + + // Support startup modes without special + if !status.Runtime { + if status.ReadUnified { + if status.WriteLegacy { + // Write both, read unified + return grafanarest.NewDualWriter(grafanarest.Mode3, legacy, storage, m.reg, gr.String()), nil + } + return storage, nil + } + if status.WriteUnified { + // Write both, read legacy + return grafanarest.NewDualWriter(grafanarest.Mode2, legacy, storage, m.reg, gr.String()), nil + } + return legacy, nil + } + + return &mangedMode3{ + service: m, + legacy: legacy, + unified: storage, + target: grafanarest.NewDualWriter(grafanarest.Mode3, legacy, storage, m.reg, gr.String()), + gr: gr, + }, nil +} + +type mangedMode3 struct { + service Service + legacy grafanarest.LegacyStorage + unified grafanarest.Storage + target grafanarest.Storage + gr schema.GroupResource +} + +func (d *mangedMode3) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) { + if d.service.ReadFromUnified(ctx, d.gr) { + return d.unified.Get(ctx, name, options) + } + return d.legacy.Get(ctx, name, options) +} + +func (d *mangedMode3) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) { + if d.service.ReadFromUnified(ctx, d.gr) { + return d.unified.List(ctx, options) + } + return d.legacy.List(ctx, options) +} + +func (d *mangedMode3) isMigrating(ctx context.Context) error { + status, ok := d.service.Status(ctx, d.gr) + if ok && status.Migrating > 0 { + return &apierrors.StatusError{ + ErrStatus: metav1.Status{ + Code: http.StatusServiceUnavailable, + Message: "the system is migrating", + }, + } + } + return nil +} + +func (d *mangedMode3) Create(ctx context.Context, in runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) { + if err := d.isMigrating(ctx); err != nil { + return nil, err + } + return d.target.Create(ctx, in, createValidation, options) +} + +func (d *mangedMode3) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) { + if err := d.isMigrating(ctx); err != nil { + return nil, false, err + } + return d.target.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options) +} + +func (d *mangedMode3) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) { + if err := d.isMigrating(ctx); err != nil { + return nil, false, err + } + return d.target.Delete(ctx, name, deleteValidation, options) +} + +// DeleteCollection overrides the behavior of the generic DualWriter and deletes from both LegacyStorage and Storage. +func (d *mangedMode3) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error) { + if err := d.isMigrating(ctx); err != nil { + return nil, err + } + return d.target.DeleteCollection(ctx, deleteValidation, options, listOptions) +} + +func (d *mangedMode3) Destroy() { + d.target.Destroy() +} + +func (d *mangedMode3) GetSingularName() string { + return d.target.GetSingularName() +} + +func (d *mangedMode3) NamespaceScoped() bool { + return d.target.NamespaceScoped() +} + +func (d *mangedMode3) New() runtime.Object { + return d.target.New() +} + +func (d *mangedMode3) NewList() runtime.Object { + return d.target.NewList() +} + +func (d *mangedMode3) ConvertToTable(ctx context.Context, object runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) { + return d.target.ConvertToTable(ctx, object, tableOptions) +} diff --git a/pkg/storage/legacysql/dualwrite/service.go b/pkg/storage/legacysql/dualwrite/service.go new file mode 100644 index 00000000000..1f231414bf5 --- /dev/null +++ b/pkg/storage/legacysql/dualwrite/service.go @@ -0,0 +1,142 @@ +package dualwrite + +import ( + "context" + "fmt" + "path/filepath" + "time" + + "github.com/prometheus/client_golang/prometheus" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/grafana/grafana/pkg/services/featuremgmt" + "github.com/grafana/grafana/pkg/setting" +) + +func ProvideService(features featuremgmt.FeatureToggles, reg prometheus.Registerer, cfg *setting.Cfg) Service { + path := "" // storage path + if cfg != nil { + path = filepath.Join(cfg.DataPath, "dualwrite.json") + } + + return &service{ + db: newFileDB(path), + reg: reg, + enabled: features.IsEnabledGlobally(featuremgmt.FlagManagedDualWriter), + // TODO: when we can "export" from legacy, this can enabled along with provisioning + // || features.IsEnabledGlobally(featuremgmt.FlagProvisioning), // required for git provisioning + } +} + +type service struct { + db statusStorage + reg prometheus.Registerer + enabled bool +} + +// The storage interface has zero business logic and simply writes values to a database +type statusStorage interface { + Get(ctx context.Context, gr schema.GroupResource) (StorageStatus, bool) + Set(ctx context.Context, status StorageStatus) error +} + +// Hardcoded list of resources that should be controlled by the database (eventually everything?) +func (m *service) ShouldManage(gr schema.GroupResource) bool { + if !m.enabled { + return false + } + switch gr.String() { + case "folders.folder.grafana.app": + return true + case "dashboards.dashboard.grafana.app": + return true + } + return false +} + +func (m *service) ReadFromUnified(ctx context.Context, gr schema.GroupResource) bool { + v, ok := m.db.Get(ctx, gr) + return ok && v.ReadUnified +} + +// Status implements Service. +func (m *service) Status(ctx context.Context, gr schema.GroupResource) (StorageStatus, bool) { + v, found := m.db.Get(ctx, gr) + if !found { + v = StorageStatus{ + Group: gr.Group, + Resource: gr.Resource, + WriteLegacy: true, + WriteUnified: true, + ReadUnified: false, + Migrated: 0, + Migrating: 0, + Runtime: true, // need to explicitly ask for not runtime + UpdateKey: 1, + } + _ = m.db.Set(ctx, v) // write the value + return v, false + } + return v, found +} + +// StartMigration implements Service. +func (m *service) StartMigration(ctx context.Context, gr schema.GroupResource, key int64) (StorageStatus, error) { + now := time.Now().UnixMilli() + v, ok := m.db.Get(ctx, gr) + if ok { + if v.Migrated > 0 { + return v, fmt.Errorf("already migrated") + } + if key != v.UpdateKey { + return v, fmt.Errorf("key mismatch") + } + if v.Migrating > 0 { + return v, fmt.Errorf("migration in progress") + } + + v.Migrating = now + v.UpdateKey++ + } else { + v = StorageStatus{ + Group: gr.Group, + Resource: gr.Resource, + Runtime: true, + WriteLegacy: true, + WriteUnified: true, + ReadUnified: false, + Migrating: now, + Migrated: 0, // timestamp + UpdateKey: 1, + } + } + err := m.db.Set(ctx, v) + return v, err +} + +// FinishMigration implements Service. +func (m *service) Update(ctx context.Context, status StorageStatus) (StorageStatus, error) { + v, ok := m.db.Get(ctx, schema.GroupResource{Group: status.Group, Resource: status.Resource}) + if !ok { + return v, fmt.Errorf("no running status") + } + if status.UpdateKey != v.UpdateKey { + return v, fmt.Errorf("key mismatch") + } + if status.Migrating > 0 { + return v, fmt.Errorf("update can not change migrating status") + } + if status.ReadUnified { + if status.Migrated == 0 { + return v, fmt.Errorf("can not read from unified before a migration") + } + if !status.WriteUnified { + return v, fmt.Errorf("must write to unified when reading from unified") + } + } + if !status.WriteLegacy && !status.WriteUnified { + return v, fmt.Errorf("must write either legacy or unified") + } + status.UpdateKey++ + return status, m.db.Set(ctx, status) +} diff --git a/pkg/storage/legacysql/dualwrite/service_test.go b/pkg/storage/legacysql/dualwrite/service_test.go new file mode 100644 index 00000000000..fc409a85fbd --- /dev/null +++ b/pkg/storage/legacysql/dualwrite/service_test.go @@ -0,0 +1,46 @@ +package dualwrite + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/grafana/grafana/pkg/services/featuremgmt" +) + +func TestService(t *testing.T) { + ctx := context.Background() + mode := ProvideService(featuremgmt.WithFeatures(), nil, nil) + + gr := schema.GroupResource{Group: "ggg", Resource: "rrr"} + status, found := mode.Status(ctx, gr) + require.False(t, found, "initially not found") + require.Equal(t, StorageStatus{ + Group: "ggg", + Resource: "rrr", + WriteLegacy: true, + WriteUnified: true, + ReadUnified: false, + Migrated: 0, + Migrating: 0, + Runtime: true, + UpdateKey: 1, + }, status, "should start with the right defaults") + + // Start migration + status, err := mode.StartMigration(ctx, gr, 1) + require.NoError(t, err) + require.Equal(t, status.UpdateKey, int64(2), "the key increased") + require.True(t, status.Migrating > 0, "migration is running") + + status.Migrated = time.Now().UnixMilli() + status.Migrating = 0 + status, err = mode.Update(ctx, status) + require.NoError(t, err) + require.Equal(t, status.UpdateKey, int64(3), "the key increased") + require.Equal(t, status.Migrating, int64(0), "done migrating") + require.True(t, status.Migrated > 0, "migration is running") +} diff --git a/pkg/storage/legacysql/dualwrite/storage_file.go b/pkg/storage/legacysql/dualwrite/storage_file.go new file mode 100644 index 00000000000..0deb01c96d7 --- /dev/null +++ b/pkg/storage/legacysql/dualwrite/storage_file.go @@ -0,0 +1,75 @@ +package dualwrite + +import ( + "context" + "encoding/json" + "os" + "sync" + + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/grafana/grafana-app-sdk/logging" +) + +// Simple file implementation -- useful while testing and not yet sure about the SQL structure! +// When a path exists, read/write it from disk +type fileDB struct { + path string + changed int64 + db map[string]StorageStatus + mu sync.RWMutex + logger logging.Logger +} + +// File implementation while testing -- values are saved in the data directory +func newFileDB(path string) *fileDB { + return &fileDB{ + db: make(map[string]StorageStatus), + path: path, + logger: logging.DefaultLogger.With("logger", "fileDB"), + } +} + +func (m *fileDB) Get(ctx context.Context, gr schema.GroupResource) (StorageStatus, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + + info, err := os.Stat(m.path) + if err == nil && info.ModTime().UnixMilli() != m.changed { + v, err := os.ReadFile(m.path) + if err == nil { + err = json.Unmarshal(v, &m.db) + m.changed = info.ModTime().UnixMilli() + } + if err != nil { + m.logger.Warn("error reading filedb", "err", err) + } + } + + v, ok := m.db[gr.String()] + return v, ok +} + +func (m *fileDB) Set(ctx context.Context, status StorageStatus) error { + m.mu.Lock() + defer m.mu.Unlock() + + gr := schema.GroupResource{ + Group: status.Group, + Resource: status.Resource, + } + m.db[gr.String()] = status + + if m.path != "" { + data, err := json.MarshalIndent(m.db, "", " ") + if err != nil { + return err + } + err = os.WriteFile(m.path, data, 0644) + if err != nil { + return err + } + } + + return nil +} diff --git a/pkg/storage/legacysql/dualwrite/storage_sql_mig.go b/pkg/storage/legacysql/dualwrite/storage_sql_mig.go new file mode 100644 index 00000000000..3bb4650ff78 --- /dev/null +++ b/pkg/storage/legacysql/dualwrite/storage_sql_mig.go @@ -0,0 +1,20 @@ +package dualwrite + +import "github.com/grafana/grafana/pkg/services/sqlstore/migrator" + +func AddUnifiedStatusMigrations(mg *migrator.Migrator) { + resourceStorageStatus := migrator.Table{ + Name: "resource_storage_status", + Columns: []*migrator.Column{ + {Name: "group", Type: migrator.DB_NVarchar, Length: 190, Nullable: false}, + {Name: "resource", Type: migrator.DB_NVarchar, Length: 190, Nullable: false}, + {Name: "migrated", Type: migrator.DB_BigInt, Nullable: false}, // Timestamp when we can start trusting unified storage + {Name: "migrating", Type: migrator.DB_BigInt, Nullable: false}, // Actively running a migration (start timestamp) + {Name: "update_key", Type: migrator.DB_BigInt, Nullable: false}, // optimistic lock key -- required for update + }, + Indices: []*migrator.Index{ + {Cols: []string{"group", "resource", "namespace"}, Type: migrator.UniqueIndex}, + }, + } + mg.AddMigration("create resource_storage_status table", migrator.NewAddTableMigration(resourceStorageStatus)) +} diff --git a/pkg/storage/legacysql/dualwrite/types.go b/pkg/storage/legacysql/dualwrite/types.go new file mode 100644 index 00000000000..5dc4401db6b --- /dev/null +++ b/pkg/storage/legacysql/dualwrite/types.go @@ -0,0 +1,52 @@ +package dualwrite + +import ( + "context" + + "k8s.io/apimachinery/pkg/runtime/schema" + + grafanarest "github.com/grafana/grafana/pkg/apiserver/rest" +) + +// For *legacy* services, this will indicate if we have transitioned to Unified storage yet +type StorageStatus struct { + Group string `json:"group" xorm:"group"` + Resource string `json:"resource" xorm:"resource"` + WriteLegacy bool `json:"write_legacy" xorm:"write_legacy"` + WriteUnified bool `json:"write_unified" xorm:"write_unified"` + + // Unified is the primary source (legacy may be secondary) + ReadUnified bool `json:"read_unified" xorm:"read_unified"` + + // Timestamp when a migration finished + Migrated int64 `json:"migrated" xorm:"migrated"` + + // Timestamp when a migration *started* this should be cleared when finished + // While migrating all write commands will be unavaliable + Migrating int64 `json:"migrating" xorm:"migrating"` + + // When false, the behavior will not change at runtime + Runtime bool `json:"runtime" xorm:"runtime"` + + // UpdateKey used for optimistic locking -- requests to change the status must match previous value + UpdateKey int64 `json:"update_key" xorm:"update_key"` +} + +type Service interface { + ShouldManage(gr schema.GroupResource) bool + + // Create a managed k8s storage instance + NewStorage(gr schema.GroupResource, legacy grafanarest.LegacyStorage, storage grafanarest.Storage) (grafanarest.Storage, error) + + // Check if the dual writes is reading from unified storage (mode3++) + ReadFromUnified(ctx context.Context, gr schema.GroupResource) bool + + // Get status details for a Group/Resource + Status(ctx context.Context, gr schema.GroupResource) (StorageStatus, bool) + + // Start a migration process (writes will be locked) + StartMigration(ctx context.Context, gr schema.GroupResource, key int64) (StorageStatus, error) + + // change the status (finish migration etc) + Update(ctx context.Context, status StorageStatus) (StorageStatus, error) +}