Unified Storage: create kind_version table migration, add SQL and fix db (#87977)

* fix database interfaces * add queries * fix queries * fix linters * add owner to imported go library * remove unused funcs * run go work sync * improve critical section fix in data race fix * fix linters * remove sync * fix typo * improve data embedding * fix linters * fix migration * remove unnecessary comments * fix linters * improve SQL templates readability * remove group_version from kind_version for consistency in History method * add created_at and updated_at columns to kind_version table
2024-05-22 11:59:40 -03:00
parent 6744dae806
commit 8b02b6b76a
20 changed files with 742 additions and 5 deletions
@@ -0,0 +1,59 @@
+package dbimpl
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	entitydb "github.com/grafana/grafana/pkg/services/store/entity/db"
+)
+
+func NewDB(d *sql.DB, driverName string) entitydb.DB {
+	return sqldb{
+		DB:         d,
+		driverName: driverName,
+	}
+}
+
+type sqldb struct {
+	*sql.DB
+	driverName string
+}
+
+func (d sqldb) DriverName() string {
+	return d.driverName
+}
+
+func (d sqldb) BeginTx(ctx context.Context, opts *sql.TxOptions) (entitydb.Tx, error) {
+	t, err := d.DB.BeginTx(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	return tx{
+		Tx: t,
+	}, nil
+}
+
+func (d sqldb) WithTx(ctx context.Context, opts *sql.TxOptions, f entitydb.TxFunc) error {
+	t, err := d.BeginTx(ctx, opts)
+	if err != nil {
+		return fmt.Errorf("begin tx: %w", err)
+	}
+
+	if err := f(ctx, t); err != nil {
+		if rollbackErr := t.Rollback(); rollbackErr != nil {
+			return fmt.Errorf("tx err: %w; rollback err: %w", err, rollbackErr)
+		}
+		return fmt.Errorf("tx err: %w", err)
+	}
+
+	if err = t.Commit(); err != nil {
+		return fmt.Errorf("commit err: %w", err)
+	}
+
+	return nil
+}
+
+type tx struct {
+	*sql.Tx
+}
@@ -0,0 +1,54 @@
+package dbimpl
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/require"
+
+	entitydb "github.com/grafana/grafana/pkg/services/store/entity/db"
+)
+
+func newCtx(t *testing.T) context.Context {
+	t.Helper()
+
+	d, ok := t.Deadline()
+	if !ok {
+		// provide a default timeout for tests
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		t.Cleanup(cancel)
+
+		return ctx
+	}
+
+	ctx, cancel := context.WithDeadline(context.Background(), d)
+	t.Cleanup(cancel)
+
+	return ctx
+}
+
+func TestDB_WithTx(t *testing.T) {
+	t.Parallel()
+
+	newTxFunc := func(err error) entitydb.TxFunc {
+		return func(context.Context, entitydb.Tx) error {
+			return err
+		}
+	}
+
+	t.Run("happy path", func(t *testing.T) {
+		t.Parallel()
+
+		sqldb, mock, err := sqlmock.New()
+		require.NoError(t, err)
+		db := NewDB(sqldb, "sqlmock")
+
+		mock.ExpectBegin()
+		mock.ExpectCommit()
+		err = db.WithTx(newCtx(t), nil, newTxFunc(nil))
+
+		require.NoError(t, err)
+	})
+}
@@ -4,6 +4,10 @@ import (
 	"fmt"

 	"github.com/dlmiddlecote/sqlstats"
+	"github.com/jmoiron/sqlx"
+	"github.com/prometheus/client_golang/prometheus"
+	"xorm.io/xorm"
+
 	"github.com/grafana/grafana/pkg/infra/db"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/tracing"
@@ -12,9 +16,6 @@ import (
 	entitydb "github.com/grafana/grafana/pkg/services/store/entity/db"
 	"github.com/grafana/grafana/pkg/services/store/entity/db/migrations"
 	"github.com/grafana/grafana/pkg/setting"
-	"github.com/jmoiron/sqlx"
-	"github.com/prometheus/client_golang/prometheus"
-	"xorm.io/xorm"
 )

 var _ entitydb.EntityDBInterface = (*EntityDB)(nil)
@@ -128,3 +129,14 @@ func (db *EntityDB) GetSession() (*session.SessionDB, error) {
 func (db *EntityDB) GetCfg() *setting.Cfg {
 	return db.cfg
 }
+
+func (db *EntityDB) GetDB() (entitydb.DB, error) {
+	engine, err := db.GetEngine()
+	if err != nil {
+		return nil, err
+	}
+
+	ret := NewDB(engine.DB().DB, engine.Dialect().DriverName())
+
+	return ret, nil
+}
@@ -187,6 +187,20 @@ func initEntityTables(mg *migrator.Migrator) string {
 		},
 	})

+	tables = append(tables, migrator.Table{
+		Name: "kind_version",
+		Columns: []*migrator.Column{
+			{Name: "group", Type: migrator.DB_NVarchar, Length: 190, Nullable: false},
+			{Name: "resource", Type: migrator.DB_NVarchar, Length: 190, Nullable: false},
+			{Name: "resource_version", Type: migrator.DB_BigInt, Nullable: false},
+			{Name: "created_at", Type: migrator.DB_BigInt, Nullable: false},
+			{Name: "updated_at", Type: migrator.DB_BigInt, Nullable: false},
+		},
+		Indices: []*migrator.Index{
+			{Cols: []string{"group", "resource"}, Type: migrator.UniqueIndex},
+		},
+	})
+
 	// Initialize all tables
 	for t := range tables {
 		mg.AddMigration("drop table "+tables[t].Name, migrator.NewDropTableMigration(tables[t].Name))
@@ -1,16 +1,58 @@
 package db

 import (
+	"context"
+	"database/sql"
+
 	"xorm.io/xorm"

-	// "github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/services/sqlstore/session"
 	"github.com/grafana/grafana/pkg/setting"
 )

+const (
+	DriverPostgres = "postgres"
+	DriverMySQL    = "mysql"
+	DriverSQLite3  = "sqlite3"
+)
+
+// EntityDBInterface provides access to a database capable of supporting the
+// Entity Server.
 type EntityDBInterface interface {
 	Init() error
+	GetCfg() *setting.Cfg
+	GetDB() (DB, error)
+
+	// TODO: deprecate.
 	GetSession() (*session.SessionDB, error)
 	GetEngine() (*xorm.Engine, error)
-	GetCfg() *setting.Cfg
+}
+
+// DB is a thin abstraction on *sql.DB to allow mocking to provide better unit
+// testing. We purposefully hide database operation methods that would use
+// context.Background().
+type DB interface {
+	ContextExecer
+	BeginTx(context.Context, *sql.TxOptions) (Tx, error)
+	WithTx(context.Context, *sql.TxOptions, TxFunc) error
+	PingContext(context.Context) error
+	Stats() sql.DBStats
+	DriverName() string
+}
+
+type TxFunc = func(context.Context, Tx) error
+
+type Tx interface {
+	ContextExecer
+	Exec(query string, args ...any) (sql.Result, error)
+	Query(query string, args ...any) (*sql.Rows, error)
+	QueryRow(query string, args ...any) *sql.Row
+	Commit() error
+	Rollback() error
+}
+
+type ContextExecer interface {
+	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
+	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
 }