Zanzana: database migrations (#89390)

* Zanana: Use grafana migrations to run openFGA migration files and initilize store.

* Add feature toggle

* Zanzana: return noop client if feature toggle is disabled

packages/grafana-data/src/types/featureToggles.gen.ts

@@ -197,4 +197,5 @@ export interface FeatureToggles {
   openSearchBackendFlowEnabled?: boolean;
   ssoSettingsLDAP?: boolean;
   databaseReadReplica?: boolean;
+  zanzana?: boolean;
 }

pkg/services/authz/zanzana.go

@@ -12,6 +12,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"github.com/grafana/grafana/pkg/infra/db"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/tracing"
 	"github.com/grafana/grafana/pkg/services/authz/zanzana"
@@ -22,9 +23,16 @@ import (
 
 type ZanzanaClient interface{}
 
-func ProvideZanzana(cfg *setting.Cfg) (ZanzanaClient, error) {
+// ProvideZanzana used to register ZanzanaClient.
-	var client *zanzana.Client
+// It will also start an embedded ZanzanaSever if mode is set to "embedded".
+func ProvideZanzana(cfg *setting.Cfg, db db.DB, features featuremgmt.FeatureToggles) (ZanzanaClient, error) {
+	if !features.IsEnabledGlobally(featuremgmt.FlagZanzana) {
+		return zanzana.NoopClient{}, nil
+	}
+
+	logger := log.New("zanzana")
 
+	var client *zanzana.Client
 	switch cfg.Zanzana.Mode {
 	case setting.ZanzanaModeClient:
 		conn, err := grpc.NewClient(cfg.Zanzana.Addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
@@ -33,10 +41,16 @@ func ProvideZanzana(cfg *setting.Cfg) (ZanzanaClient, error) {
 		}
 		client = zanzana.NewClient(openfgav1.NewOpenFGAServiceClient(conn))
 	case setting.ZanzanaModeEmbedded:
-		srv, err := zanzana.NewServer(zanzana.NewStore())
+		store, err := zanzana.NewEmbeddedStore(cfg, db, logger)
+		if err != nil {
+			return nil, fmt.Errorf("failed to start zanzana: %w", err)
+		}
+
+		srv, err := zanzana.NewServer(store, logger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to start zanzana: %w", err)
 		}
+
 		channel := &inprocgrpc.Channel{}
 		openfgav1.RegisterOpenFGAServiceServer(channel, srv)
 		client = zanzana.NewClient(openfgav1.NewOpenFGAServiceClient(channel))
@@ -77,7 +91,12 @@ type Zanzana struct {
 }
 
 func (z *Zanzana) start(ctx context.Context) error {
-	srv, err := zanzana.NewServer(zanzana.NewStore())
+	store, err := zanzana.NewStore(z.cfg, z.logger)
+	if err != nil {
+		return fmt.Errorf("failed to initilize zanana store: %w", err)
+	}
+
+	srv, err := zanzana.NewServer(store, z.logger)
 	if err != nil {
 		return fmt.Errorf("failed to start zanzana: %w", err)
 	}

pkg/services/authz/zanzana/client.go

@@ -12,3 +12,5 @@ type Client struct {
 func NewClient(c openfgav1.OpenFGAServiceClient) *Client {
 	return &Client{c}
 }
+
+type NoopClient struct{}

pkg/services/authz/zanzana/logger.go

@@ -10,12 +10,10 @@ import (
 
 // zanzanaLogger is a grafana logger wrapper compatible with OpenFGA logger interface
 type zanzanaLogger struct {
-	logger *log.ConcreteLogger
+	logger log.Logger
 }
 
-func newZanzanaLogger() *zanzanaLogger {
+func newZanzanaLogger(logger log.Logger) *zanzanaLogger {
-	logger := log.New("openfga-server")
-
 	return &zanzanaLogger{
 		logger: logger,
 	}
@@ -23,7 +21,8 @@ func newZanzanaLogger() *zanzanaLogger {
 
 // Simple converter for zap logger fields
 func zapFieldsToArgs(fields []zap.Field) []any {
-	args := make([]any, 0)
+	// We need to pre-allocated space for key and value
+	args := make([]any, 0, len(fields)*2)
 	for _, f := range fields {
 		args = append(args, f.Key)
 		if f.Interface != nil {

pkg/services/authz/zanzana/server.go

@@ -3,13 +3,15 @@ package zanzana
 import (
 	"github.com/openfga/openfga/pkg/server"
 	"github.com/openfga/openfga/pkg/storage"
+
+	"github.com/grafana/grafana/pkg/infra/log"
 )
 
-func NewServer(store storage.OpenFGADatastore) (*server.Server, error) {
+func NewServer(store storage.OpenFGADatastore, logger log.Logger) (*server.Server, error) {
 	// FIXME(kalleep): add support for more options, configure logging, tracing etc
 	opts := []server.OpenFGAServiceV1Option{
 		server.WithDatastore(store),
-		server.WithLogger(newZanzanaLogger()),
+		server.WithLogger(newZanzanaLogger(logger)),
 	}
 
 	// FIXME(kalleep): Interceptors

pkg/services/authz/zanzana/store.go

@@ -1,13 +1,214 @@
 package zanzana
 
 import (
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/openfga/openfga/assets"
 	"github.com/openfga/openfga/pkg/storage"
 	"github.com/openfga/openfga/pkg/storage/memory"
+	"github.com/openfga/openfga/pkg/storage/mysql"
+	"github.com/openfga/openfga/pkg/storage/postgres"
+	"github.com/openfga/openfga/pkg/storage/sqlcommon"
+	"xorm.io/xorm"
+
+	"github.com/grafana/grafana/pkg/infra/db"
+	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/services/sqlstore"
+	"github.com/grafana/grafana/pkg/services/sqlstore/migrator"
+	"github.com/grafana/grafana/pkg/setting"
 )
 
-// FIXME(kalleep): Add support for postgres, mysql and sqlite data stores.
+// FIXME(kalleep): Add sqlite data store.
-// Postgres and mysql is already implmented by openFGA so we just need to hook up migartions for them.
 // There is no support for sqlite atm but we are working on adding it: https://github.com/openfga/openfga/pull/1615
-func NewStore() storage.OpenFGADatastore {
+func NewStore(cfg *setting.Cfg, logger log.Logger) (storage.OpenFGADatastore, error) {
-	return memory.New()
+	grafanaDBCfg, zanzanaDBCfg, err := parseConfig(cfg, logger)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse database config: %w", err)
+	}
+
+	switch grafanaDBCfg.Type {
+	case migrator.SQLite:
+		return memory.New(), nil
+	case migrator.MySQL:
+		// For mysql we need to pass parseTime parameter in connection string
+		connStr := grafanaDBCfg.ConnectionString + "&parseTime=true"
+		if err := runMigrations(cfg, migrator.MySQL, connStr, assets.MySQLMigrationDir); err != nil {
+			return nil, fmt.Errorf("failed to run migrations: %w", err)
+		}
+		return mysql.New(connStr, zanzanaDBCfg)
+	case migrator.Postgres:
+		if err := runMigrations(cfg, migrator.Postgres, grafanaDBCfg.ConnectionString, assets.PostgresMigrationDir); err != nil {
+			return nil, fmt.Errorf("failed to run migrations: %w", err)
+		}
+		return postgres.New(grafanaDBCfg.ConnectionString, zanzanaDBCfg)
+	}
+
+	// Should never happen
+	return nil, fmt.Errorf("unsupported database engine: %s", grafanaDBCfg.Type)
+}
+
+func NewEmbeddedStore(cfg *setting.Cfg, db db.DB, logger log.Logger) (storage.OpenFGADatastore, error) {
+	grafanaDBCfg, zanzanaDBCfg, err := parseConfig(cfg, logger)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse database config: %w", err)
+	}
+
+	m := migrator.NewMigrator(db.GetEngine(), cfg)
+
+	switch grafanaDBCfg.Type {
+	case migrator.SQLite:
+		// FIXME(kalleep): At the moment sqlite is not a supported data store.
+		// So we just return in memory store for now.
+		return memory.New(), nil
+	case migrator.MySQL:
+		if err := runMigrationsWithMigrator(m, cfg, assets.MySQLMigrationDir); err != nil {
+			return nil, fmt.Errorf("failed to run migrations: %w", err)
+		}
+
+		// For mysql we need to pass parseTime parameter in connection string
+		return mysql.New(grafanaDBCfg.ConnectionString+"&parseTime=true", zanzanaDBCfg)
+	case migrator.Postgres:
+		if err := runMigrationsWithMigrator(m, cfg, assets.PostgresMigrationDir); err != nil {
+			return nil, fmt.Errorf("failed to run migrations: %w", err)
+		}
+
+		return postgres.New(grafanaDBCfg.ConnectionString, zanzanaDBCfg)
+	}
+
+	// Should never happen
+	return nil, fmt.Errorf("unsupported database engine: %s", db.GetDialect().DriverName())
+}
+
+func parseConfig(cfg *setting.Cfg, logger log.Logger) (*sqlstore.DatabaseConfig, *sqlcommon.Config, error) {
+	sec := cfg.Raw.Section("database")
+	grafanaDBCfg, err := sqlstore.NewDatabaseConfig(cfg, nil)
+	if err != nil {
+		return nil, nil, nil
+	}
+
+	zanzanaDBCfg := &sqlcommon.Config{
+		Logger: newZanzanaLogger(logger),
+		// MaxTuplesPerWriteField: 0,
+		// MaxTypesPerModelField:  0,
+		MaxOpenConns:    grafanaDBCfg.MaxOpenConn,
+		MaxIdleConns:    grafanaDBCfg.MaxIdleConn,
+		ConnMaxLifetime: time.Duration(grafanaDBCfg.ConnMaxLifetime) * time.Second,
+		ExportMetrics:   sec.Key("instrument_queries").MustBool(false),
+	}
+
+	return grafanaDBCfg, zanzanaDBCfg, nil
+}
+
+func runMigrations(cfg *setting.Cfg, typ, connStr, path string) error {
+	engine, err := xorm.NewEngine(typ, connStr)
+	if err != nil {
+		return fmt.Errorf("failed to parse database config: %w", err)
+	}
+
+	m := migrator.NewMigrator(engine, cfg)
+	m.AddCreateMigration()
+
+	return runMigrationsWithMigrator(m, cfg, path)
+}
+
+func runMigrationsWithMigrator(m *migrator.Migrator, cfg *setting.Cfg, path string) error {
+	migrations, err := getMigrations(path)
+	if err != nil {
+		return err
+	}
+
+	for _, mig := range migrations {
+		m.AddMigration(mig.name, mig.migration)
+	}
+
+	sec := cfg.Raw.Section("database")
+	return m.Start(
+		sec.Key("migration_locking").MustBool(true),
+		sec.Key("locking_attempt_timeout_sec").MustInt(),
+	)
+}
+
+type migration struct {
+	name      string
+	migration migrator.Migration
+}
+
+func getMigrations(path string) ([]migration, error) {
+	entries, err := assets.EmbedMigrations.ReadDir(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read migration dir: %w", err)
+	}
+
+	// parseStatements extracts statements from a sql file so we can execute
+	// them as separate migrations. OpenFGA uses Goose as their migration egine
+	// and Goose uses a single sql file for both up and down migrations.
+	// Grafana only supports up migration so we strip out the down migration
+	// and parse each individual statement
+	parseStatements := func(data []byte) ([]string, error) {
+		scripts := strings.Split(strings.TrimPrefix(string(data), "-- +goose Up"), "-- +goose Down")
+		if len(scripts) != 2 {
+			return nil, errors.New("malformed migration file")
+		}
+
+		// We assume that up migrations are always before down migrations
+		parts := strings.SplitAfter(scripts[0], ";")
+		stmts := make([]string, 0, len(parts))
+		for _, p := range parts {
+			p = strings.TrimSpace(p)
+			if p != "" {
+				stmts = append(stmts, p)
+			}
+		}
+
+		return stmts, nil
+	}
+
+	formatName := func(name string) string {
+		// Each migration file start with XXX where X is a number.
+		// We remove that part and prefix each migration with "zanzana".
+		return strings.TrimSuffix("zanzana"+name[3:], ".sql")
+	}
+
+	migrations := make([]migration, 0, len(entries))
+	for _, e := range entries {
+		data, err := assets.EmbedMigrations.ReadFile(path + "/" + e.Name())
+		if err != nil {
+			return nil, fmt.Errorf("failed to read migration file: %w", err)
+		}
+
+		stmts, err := parseStatements(data)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse migration: %w", err)
+		}
+
+		migrations = append(migrations, migration{
+			name:      formatName(e.Name()),
+			migration: &rawMigration{stmts: stmts},
+		})
+	}
+
+	return migrations, nil
+}
+
+var _ migrator.CodeMigration = (*rawMigration)(nil)
+
+type rawMigration struct {
+	stmts []string
+	migrator.MigrationBase
+}
+
+func (m *rawMigration) Exec(sess *xorm.Session, migrator *migrator.Migrator) error {
+	for _, stmt := range m.stmts {
+		if _, err := sess.Exec(stmt); err != nil {
+			return fmt.Errorf("failed to run migration: %w", err)
+		}
+	}
+	return nil
+}
+
+func (m *rawMigration) SQL(dialect migrator.Dialect) string {
+	return strings.Join(m.stmts, "\n")
 }

pkg/services/featuremgmt/registry.go

@@ -1343,6 +1343,14 @@ var (
 			Owner:       grafanaBackendServicesSquad,
 			Expression:  "false", // enabled by default
 		},
+		{
+			Name:              "zanzana",
+			Description:       "Use openFGA as authorization engine.",
+			Stage:             FeatureStageExperimental,
+			Owner:             identityAccessTeam,
+			HideFromDocs:      true,
+			HideFromAdminPage: true,
+		},
 	}
 )
 

pkg/services/featuremgmt/toggles_gen.csv

@@ -178,3 +178,4 @@ authZGRPCServer,experimental,@grafana/identity-access-team,false,false,false
 openSearchBackendFlowEnabled,preview,@grafana/aws-datasources,false,false,false
 ssoSettingsLDAP,experimental,@grafana/identity-access-team,false,false,false
 databaseReadReplica,experimental,@grafana/grafana-backend-services-squad,false,false,false
+zanzana,experimental,@grafana/identity-access-team,false,false,false

pkg/services/featuremgmt/toggles_gen.go

@@ -722,4 +722,8 @@ const (
 	// FlagDatabaseReadReplica
 	// Use a read replica for some database queries.
 	FlagDatabaseReadReplica = "databaseReadReplica"
+
+	// FlagZanzana
+	// Use openFGA as authorization engine.
+	FlagZanzana = "zanzana"
 )

pkg/services/featuremgmt/toggles_gen.json

@@ -2305,6 +2305,20 @@
         "stage": "experimental",
         "codeowner": "@grafana/hosted-grafana-team"
       }
+    },
+    {
+      "metadata": {
+        "name": "zanzana",
+        "resourceVersion": "1718787304727",
+        "creationTimestamp": "2024-06-19T08:55:04Z"
+      },
+      "spec": {
+        "description": "Use openFGA as authorization engine.",
+        "stage": "experimental",
+        "codeowner": "@grafana/identity-access-team",
+        "hideFromAdminPage": true,
+        "hideFromDocs": true
+      }
     }
   ]
 }