Storage: Add ManagedDualWriter mode (#100231)

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;

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,

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

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 {

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",

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

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"

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",

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)
+}

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)
+}

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")
+}

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
+}

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))
+}

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)
+}