mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-29 07:49:14 +01:00
[management] Remove the MySQL store engine (#7914)
This commit is contained in:
@@ -152,7 +152,6 @@ func (am *DefaultAccountManager) runAccountDeletionHooks(ctx context.Context, ac
|
||||
func isUniqueConstraintError(err error) bool {
|
||||
switch {
|
||||
case strings.Contains(err.Error(), "(SQLSTATE 23505)"),
|
||||
strings.Contains(err.Error(), "Error 1062 (23000)"),
|
||||
strings.Contains(err.Error(), "UNIQUE constraint failed"):
|
||||
return true
|
||||
|
||||
|
||||
@@ -3759,7 +3759,7 @@ func setupNetworkMapTest(t *testing.T) (*DefaultAccountManager, *update_channel.
|
||||
|
||||
// peerUpdateTimeout bounds how long peerShouldReceiveUpdate and its outer
|
||||
// wrappers wait for an expected update message. Sized for slow CI runners
|
||||
// (MySQL, FreeBSD, loaded sqlite) where the channel publish can take
|
||||
// (FreeBSD, loaded sqlite) where the channel publish can take
|
||||
// seconds. Only runs down on failure; passing tests return immediately
|
||||
// when the channel delivers.
|
||||
const peerUpdateTimeout = 5 * time.Second
|
||||
|
||||
@@ -429,10 +429,6 @@ func Test_SyncStatusRace(t *testing.T) {
|
||||
if os.Getenv("NETBIRD_STORE_ENGINE") == "postgres" {
|
||||
t.Skip("Skipping on CI and Postgres store")
|
||||
}
|
||||
|
||||
if os.Getenv("NETBIRD_STORE_ENGINE") == "mysql" {
|
||||
t.Skip("Skipping on CI and MySQL store")
|
||||
}
|
||||
}
|
||||
for i := 0; i < 500; i++ {
|
||||
t.Run(fmt.Sprintf("TestRun-%d", i), func(t *testing.T) {
|
||||
|
||||
@@ -19,19 +19,11 @@ import (
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
func GetColumnName(db *gorm.DB, column string) string {
|
||||
if db.Name() == "mysql" {
|
||||
return fmt.Sprintf("`%s`", column)
|
||||
}
|
||||
return column
|
||||
}
|
||||
|
||||
// MigrateFieldFromGobToJSON migrates a column from Gob encoding to JSON encoding.
|
||||
// T is the type of the model that contains the field to be migrated.
|
||||
// S is the type of the field to be migrated.
|
||||
func MigrateFieldFromGobToJSON[T any, S any](ctx context.Context, db *gorm.DB, fieldName string) error {
|
||||
orgColumnName := fieldName
|
||||
oldColumnName := GetColumnName(db, orgColumnName)
|
||||
oldColumnName := fieldName
|
||||
newColumnName := fieldName + "_tmp"
|
||||
|
||||
var model T
|
||||
@@ -86,7 +78,7 @@ func MigrateFieldFromGobToJSON[T any, S any](ctx context.Context, db *gorm.DB, f
|
||||
for _, row := range rows {
|
||||
var field S
|
||||
|
||||
str, ok := row[orgColumnName].(string)
|
||||
str, ok := row[oldColumnName].(string)
|
||||
if !ok {
|
||||
return fmt.Errorf("type assertion failed")
|
||||
}
|
||||
@@ -125,8 +117,7 @@ func MigrateFieldFromGobToJSON[T any, S any](ctx context.Context, db *gorm.DB, f
|
||||
// MigrateNetIPFieldFromBlobToJSON migrates a Net IP column from Blob encoding to JSON encoding.
|
||||
// T is the type of the model that contains the field to be migrated.
|
||||
func MigrateNetIPFieldFromBlobToJSON[T any](ctx context.Context, db *gorm.DB, fieldName string, indexName string) error {
|
||||
orgColumnName := fieldName
|
||||
oldColumnName := GetColumnName(db, orgColumnName)
|
||||
oldColumnName := fieldName
|
||||
newColumnName := fieldName + "_tmp"
|
||||
|
||||
var model T
|
||||
@@ -178,7 +169,7 @@ func MigrateNetIPFieldFromBlobToJSON[T any](ctx context.Context, db *gorm.DB, fi
|
||||
|
||||
for _, row := range rows {
|
||||
var blobValue string
|
||||
if columnValue := row[orgColumnName]; columnValue != nil {
|
||||
if columnValue := row[oldColumnName]; columnValue != nil {
|
||||
value, ok := columnValue.(string)
|
||||
if !ok {
|
||||
return fmt.Errorf("type assertion failed")
|
||||
@@ -225,8 +216,7 @@ func MigrateNetIPFieldFromBlobToJSON[T any](ctx context.Context, db *gorm.DB, fi
|
||||
}
|
||||
|
||||
func MigrateSetupKeyToHashedSetupKey[T any](ctx context.Context, db *gorm.DB) error {
|
||||
orgColumnName := "key"
|
||||
oldColumnName := GetColumnName(db, orgColumnName)
|
||||
oldColumnName := "key"
|
||||
newColumnName := "key_secret"
|
||||
|
||||
var model T
|
||||
@@ -268,7 +258,7 @@ func MigrateSetupKeyToHashedSetupKey[T any](ctx context.Context, db *gorm.DB) er
|
||||
for _, row := range rows {
|
||||
|
||||
var plainKey string
|
||||
if columnValue := row[orgColumnName]; columnValue != nil {
|
||||
if columnValue := row[oldColumnName]; columnValue != nil {
|
||||
value, ok := columnValue.(string)
|
||||
if !ok {
|
||||
return fmt.Errorf("type assertion failed")
|
||||
@@ -394,33 +384,13 @@ func CreateIndexIfNotExists[T any](ctx context.Context, db *gorm.DB, indexName s
|
||||
return fmt.Errorf("failed to parse model schema: %w", err)
|
||||
}
|
||||
tableName := stmt.Schema.Table
|
||||
dialect := db.Name()
|
||||
|
||||
if db.Migrator().HasIndex(&model, indexName) {
|
||||
log.WithContext(ctx).Infof("index %s already exists on table %s", indexName, tableName)
|
||||
return nil
|
||||
}
|
||||
|
||||
var columnClause string
|
||||
if dialect == "mysql" {
|
||||
var withLength []string
|
||||
for _, col := range columns {
|
||||
quotedCol := fmt.Sprintf("`%s`", col)
|
||||
if col == "ip" || col == "dns_label" || col == "key" {
|
||||
withLength = append(withLength, fmt.Sprintf("%s(64)", quotedCol))
|
||||
} else {
|
||||
withLength = append(withLength, quotedCol)
|
||||
}
|
||||
}
|
||||
columnClause = strings.Join(withLength, ", ")
|
||||
} else {
|
||||
columnClause = strings.Join(columns, ", ")
|
||||
}
|
||||
|
||||
createStmt := fmt.Sprintf("CREATE UNIQUE INDEX %s ON %s (%s)", indexName, tableName, columnClause)
|
||||
if dialect == "postgres" || dialect == "sqlite" {
|
||||
createStmt = strings.Replace(createStmt, "CREATE UNIQUE INDEX", "CREATE UNIQUE INDEX IF NOT EXISTS", 1)
|
||||
}
|
||||
createStmt := fmt.Sprintf("CREATE UNIQUE INDEX IF NOT EXISTS %s ON %s (%s)", indexName, tableName, strings.Join(columns, ", "))
|
||||
|
||||
log.WithContext(ctx).Infof("executing index creation: %s", createStmt)
|
||||
if err := db.Exec(createStmt).Error; err != nil {
|
||||
@@ -505,14 +475,6 @@ func hasForeignKey(db *gorm.DB, table, column string) bool {
|
||||
AND kcu.table_name = ?
|
||||
AND kcu.column_name = ?
|
||||
`, table, column).Scan(&count)
|
||||
case "mysql":
|
||||
db.Raw(`
|
||||
SELECT COUNT(*) FROM information_schema.key_column_usage
|
||||
WHERE table_schema = DATABASE()
|
||||
AND table_name = ?
|
||||
AND column_name = ?
|
||||
AND referenced_table_name IS NOT NULL
|
||||
`, table, column).Scan(&count)
|
||||
default: // sqlite
|
||||
type fkInfo struct {
|
||||
From string
|
||||
@@ -592,7 +554,7 @@ func RemoveDuplicatePeerKeys(ctx context.Context, db *gorm.DB) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
keyColumn := GetColumnName(db, "key")
|
||||
keyColumn := "key"
|
||||
|
||||
var duplicates []struct {
|
||||
Key string
|
||||
|
||||
@@ -33,15 +33,6 @@ func (agentNetworkSettingsMigration) TableName() string { return "agent_network_
|
||||
// bootstrap always wrote both, so such a row indicates corruption and the
|
||||
// migration fails loudly rather than leaving an empty domain to collide with
|
||||
// the unique index confusingly.
|
||||
//
|
||||
// The transaction is real only on sqlite and postgres, where DDL is
|
||||
// transactional. MySQL implicitly commits around every ALTER TABLE, so there
|
||||
// each step stands alone; what makes an interrupted run resumable on MySQL is
|
||||
// that every step is guarded by the schema state it changes — the entry check
|
||||
// fires while either legacy column remains, the adds skip existing columns,
|
||||
// the backfill and its loud-failure check run only while the legacy cluster
|
||||
// column exists (they provably completed before any drop), and each drop
|
||||
// skips what is already gone.
|
||||
func MigrateAgentNetworkSettingsToDomain(ctx context.Context, db *gorm.DB) error {
|
||||
model := &agentNetworkSettingsMigration{}
|
||||
migrator := db.Migrator()
|
||||
@@ -66,14 +57,9 @@ func MigrateAgentNetworkSettingsToDomain(ctx context.Context, db *gorm.DB) error
|
||||
}
|
||||
|
||||
if hasCluster {
|
||||
concat := "subdomain || '.' || cluster"
|
||||
if tx.Name() == "mysql" {
|
||||
concat = "CONCAT(subdomain, '.', cluster)"
|
||||
}
|
||||
res := tx.Exec(fmt.Sprintf(
|
||||
"UPDATE agent_network_settings SET domain = %s, proxy_address = cluster WHERE (domain IS NULL OR domain = '') AND cluster <> '' AND subdomain <> ''",
|
||||
concat,
|
||||
))
|
||||
res := tx.Exec(
|
||||
"UPDATE agent_network_settings SET domain = subdomain || '.' || cluster, proxy_address = cluster WHERE (domain IS NULL OR domain = '') AND cluster <> '' AND subdomain <> ''",
|
||||
)
|
||||
if res.Error != nil {
|
||||
return fmt.Errorf("backfill agent_network_settings domain: %w", res.Error)
|
||||
}
|
||||
|
||||
@@ -11,7 +11,6 @@ import (
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/mysql"
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
@@ -32,17 +31,6 @@ func setupDatabase(t *testing.T) *gorm.DB {
|
||||
var dsn string
|
||||
var cleanup func()
|
||||
switch os.Getenv("NETBIRD_STORE_ENGINE") {
|
||||
case "mysql":
|
||||
cleanup, dsn, err = testutil.CreateMysqlTestContainer()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create MySQL test container: %v", err)
|
||||
}
|
||||
|
||||
if dsn == "" {
|
||||
t.Fatal("MySQL connection string is empty, ensure the test container is running")
|
||||
}
|
||||
|
||||
db, err = gorm.Open(mysql.Open(dsn+"?charset=utf8&parseTime=True&loc=Local"), &gorm.Config{})
|
||||
case "postgres":
|
||||
cleanup, dsn, err = testutil.CreatePostgresTestContainer()
|
||||
if err != nil {
|
||||
@@ -597,9 +585,8 @@ func TestCleanupOrphanedResources_Idempotent(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCleanupOrphanedResources_SkipsWhenForeignKeyExists(t *testing.T) {
|
||||
engine := os.Getenv("NETBIRD_STORE_ENGINE")
|
||||
if engine != "postgres" && engine != "mysql" {
|
||||
t.Skip("FK constraint early-exit test requires postgres or mysql")
|
||||
if os.Getenv("NETBIRD_STORE_ENGINE") != "postgres" {
|
||||
t.Skip("FK constraint early-exit test requires postgres")
|
||||
}
|
||||
|
||||
db := setupDatabase(t)
|
||||
@@ -614,24 +601,12 @@ func TestCleanupOrphanedResources_SkipsWhenForeignKeyExists(t *testing.T) {
|
||||
require.NoError(t, db.Create(&testChildWithFK{ID: "c1", ParentID: "p1"}).Error)
|
||||
require.NoError(t, db.Create(&testChildWithFK{ID: "c2", ParentID: "p2"}).Error)
|
||||
|
||||
switch engine {
|
||||
case "postgres":
|
||||
require.NoError(t, db.Exec("ALTER TABLE test_children DROP CONSTRAINT fk_test_children_parent").Error)
|
||||
require.NoError(t, db.Exec("DELETE FROM test_parents WHERE id = ?", "p2").Error)
|
||||
require.NoError(t, db.Exec(
|
||||
"ALTER TABLE test_children ADD CONSTRAINT fk_test_children_parent "+
|
||||
"FOREIGN KEY (parent_id) REFERENCES test_parents(id) NOT VALID",
|
||||
).Error)
|
||||
case "mysql":
|
||||
require.NoError(t, db.Exec("SET FOREIGN_KEY_CHECKS = 0").Error)
|
||||
require.NoError(t, db.Exec("ALTER TABLE test_children DROP FOREIGN KEY fk_test_children_parent").Error)
|
||||
require.NoError(t, db.Exec("DELETE FROM test_parents WHERE id = ?", "p2").Error)
|
||||
require.NoError(t, db.Exec(
|
||||
"ALTER TABLE test_children ADD CONSTRAINT fk_test_children_parent "+
|
||||
"FOREIGN KEY (parent_id) REFERENCES test_parents(id)",
|
||||
).Error)
|
||||
require.NoError(t, db.Exec("SET FOREIGN_KEY_CHECKS = 1").Error)
|
||||
}
|
||||
require.NoError(t, db.Exec("ALTER TABLE test_children DROP CONSTRAINT fk_test_children_parent").Error)
|
||||
require.NoError(t, db.Exec("DELETE FROM test_parents WHERE id = ?", "p2").Error)
|
||||
require.NoError(t, db.Exec(
|
||||
"ALTER TABLE test_children ADD CONSTRAINT fk_test_children_parent "+
|
||||
"FOREIGN KEY (parent_id) REFERENCES test_parents(id) NOT VALID",
|
||||
).Error)
|
||||
|
||||
err = migration.CleanupOrphanedResources[testChildWithFK, testParent](context.Background(), db, "parent_id")
|
||||
require.NoError(t, err)
|
||||
@@ -717,8 +692,6 @@ func TestFoldCostAggregatesIntoBuckets_SkipsAlreadyMigrated(t *testing.T) {
|
||||
require.NoError(t, db.Migrator().DropTable(&agentNetworkTypes.AgentNetworkUsage{}))
|
||||
|
||||
require.NoError(t, db.AutoMigrate(&agentNetworkTypes.AgentNetworkUsage{}))
|
||||
// Timestamp must be set explicitly: a zero time.Time serialises as
|
||||
// '0000-00-00 00:00:00', which MySQL rejects under strict mode.
|
||||
require.NoError(t, db.Create(&agentNetworkTypes.AgentNetworkUsage{
|
||||
ID: "u1", AccountID: "acct-1", Model: "claude-sonnet-4-6",
|
||||
Timestamp: time.Date(2026, 5, 5, 9, 0, 0, 0, time.UTC),
|
||||
@@ -819,10 +792,9 @@ func TestMigrateAgentNetworkSettingsToDomain_FailsOnUnmigratableRow(t *testing.T
|
||||
assert.Contains(t, err.Error(), "resolve them manually", "the error must tell the operator what to do")
|
||||
}
|
||||
|
||||
// partialAgentNetworkSettings models the one non-atomic state a MySQL run can
|
||||
// be interrupted in: DDL auto-commits there, so a crash between the two legacy
|
||||
// column drops leaves subdomain behind while cluster (and the completed
|
||||
// backfill) are already committed.
|
||||
// partialAgentNetworkSettings models a schema left between the two legacy
|
||||
// column drops: subdomain is still present while cluster is gone and the
|
||||
// backfill has already completed.
|
||||
type partialAgentNetworkSettings struct {
|
||||
AccountID string `gorm:"primaryKey"`
|
||||
Subdomain string
|
||||
@@ -832,8 +804,8 @@ type partialAgentNetworkSettings struct {
|
||||
|
||||
func (partialAgentNetworkSettings) TableName() string { return "agent_network_settings" }
|
||||
|
||||
// TestMigrateAgentNetworkSettingsToDomain_ResumesAfterPartialDrop pins MySQL
|
||||
// resumability: a rerun over the interrupted state must remove the leftover
|
||||
// TestMigrateAgentNetworkSettingsToDomain_ResumesAfterPartialDrop pins
|
||||
// resumability: a rerun over the partial state must remove the leftover
|
||||
// subdomain column without re-running the backfill (the cluster column that
|
||||
// feeds it is gone) and without touching the migrated values.
|
||||
func TestMigrateAgentNetworkSettingsToDomain_ResumesAfterPartialDrop(t *testing.T) {
|
||||
|
||||
@@ -1566,8 +1566,7 @@ func Test_RegisterPeerBySetupKey(t *testing.T) {
|
||||
|
||||
func Test_RegisterPeerRollbackOnFailure(t *testing.T) {
|
||||
engine := os.Getenv("NETBIRD_STORE_ENGINE")
|
||||
if engine == "sqlite" || engine == "mysql" || engine == "" {
|
||||
// we intentionally disabled foreign keys in mysql
|
||||
if engine == "sqlite" || engine == "" {
|
||||
t.Skip("Skipping test because store is not respecting foreign keys")
|
||||
}
|
||||
if runtime.GOOS == "windows" {
|
||||
@@ -2219,10 +2218,6 @@ func Test_IsUniqueConstraintError(t *testing.T) {
|
||||
name: "PostgreSQL uniqueness error",
|
||||
engine: types.PostgresStoreEngine,
|
||||
},
|
||||
{
|
||||
name: "MySQL uniqueness error",
|
||||
engine: types.MysqlStoreEngine,
|
||||
},
|
||||
{
|
||||
name: "SQLite uniqueness error",
|
||||
engine: types.SqliteStoreEngine,
|
||||
|
||||
@@ -33,7 +33,6 @@ import (
|
||||
const (
|
||||
idQueryCondition = "id = ?"
|
||||
keyQueryCondition = "key = ?"
|
||||
mysqlKeyQueryCondition = "`key` = ?"
|
||||
accountAndIDQueryCondition = "account_id = ? and id = ?"
|
||||
accountAndAnyIDQueryCondition = "account_id = ? and (id = ? or public_id = ?)"
|
||||
accountAndPeerIDQueryCondition = "account_id = ? and peer_id = ?"
|
||||
@@ -119,13 +118,6 @@ func (s *SqlStore) pgxPool() *pgxpool.Pool {
|
||||
return s.conn.Pool(s.tx)
|
||||
}
|
||||
|
||||
func GetKeyQueryCondition(s *SqlStore) string {
|
||||
if s.conn.Engine() == db.MysqlStoreEngine {
|
||||
return mysqlKeyQueryCondition
|
||||
}
|
||||
return keyQueryCondition
|
||||
}
|
||||
|
||||
// AcquireGlobalLock acquires global lock across all the accounts and returns a function that releases the lock
|
||||
func (s *SqlStore) AcquireGlobalLock(ctx context.Context) (unlock func()) {
|
||||
log.WithContext(ctx).Tracef("acquiring global lock")
|
||||
@@ -174,15 +166,6 @@ func NewPostgresqlStore(ctx context.Context, dsn string, metrics telemetry.AppMe
|
||||
return newStore(ctx, conn, metrics, skipMigration)
|
||||
}
|
||||
|
||||
// NewMysqlStore creates a new MySQL store.
|
||||
func NewMysqlStore(ctx context.Context, dsn string, metrics telemetry.AppMetrics, skipMigration bool) (*SqlStore, error) {
|
||||
conn, err := db.OpenMysql(ctx, dsn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return newStore(ctx, conn, metrics, skipMigration)
|
||||
}
|
||||
|
||||
// NewSqliteStoreFromFileStore restores a store from FileStore and stores SQLite DB in the file located in datadir.
|
||||
func NewSqliteStoreFromFileStore(ctx context.Context, fileStore *FileStore, dataDir string, metrics telemetry.AppMetrics, skipMigration bool) (*SqlStore, error) {
|
||||
store, err := NewSqliteStore(ctx, dataDir, metrics, skipMigration)
|
||||
@@ -240,11 +223,6 @@ func NewPostgresqlStoreForTests(ctx context.Context, dsn string, metrics telemet
|
||||
return newStore(ctx, conn, metrics, skipMigration)
|
||||
}
|
||||
|
||||
// NewMysqlStoreFromSqlStore restores a store from SqlStore and stores MySQL DB.
|
||||
func NewMysqlStoreFromSqlStore(ctx context.Context, sqliteStore *SqlStore, dsn string, metrics telemetry.AppMetrics) (*SqlStore, error) {
|
||||
return newMysqlStoreFromSqlStore(ctx, sqliteStore, dsn, metrics, false)
|
||||
}
|
||||
|
||||
// seedFromSqliteStore copies the installation ID and the accounts of the
|
||||
// sqlite seed store into a freshly created engine store.
|
||||
func seedFromSqliteStore(ctx context.Context, store, sqliteStore *SqlStore) error {
|
||||
@@ -259,20 +237,6 @@ func seedFromSqliteStore(ctx context.Context, store, sqliteStore *SqlStore) erro
|
||||
return nil
|
||||
}
|
||||
|
||||
func newMysqlStoreFromSqlStore(ctx context.Context, sqliteStore *SqlStore, dsn string, metrics telemetry.AppMetrics, skipMigration bool) (*SqlStore, error) {
|
||||
store, err := NewMysqlStore(ctx, dsn, metrics, skipMigration)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := seedFromSqliteStore(ctx, store, sqliteStore); err != nil {
|
||||
_ = store.Close(ctx)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return store, nil
|
||||
}
|
||||
|
||||
// ExecuteInTransaction runs operation in a transaction. A store that is already
|
||||
// bound to one joins it instead of opening a second, independent transaction.
|
||||
func (s *SqlStore) ExecuteInTransaction(ctx context.Context, operation func(store Store) error) error {
|
||||
|
||||
@@ -1084,10 +1084,8 @@ func (s *SqlStore) SaveAccountSettings(ctx context.Context, accountID string, se
|
||||
return status.Errorf(status.Internal, "failed to save account settings to store")
|
||||
}
|
||||
|
||||
// MySQL reports RowsAffected=0 for no-op updates where values don't change,
|
||||
// unlike SQLite/Postgres which report matched rows. Skip the check since the
|
||||
// caller (UpdateAccountSettings) already verified the account exists via
|
||||
// GetAccountSettings with LockingStrengthUpdate.
|
||||
// No RowsAffected check: the caller (UpdateAccountSettings) already verified
|
||||
// the account exists via GetAccountSettings with LockingStrengthUpdate.
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package store
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -69,10 +68,6 @@ func TestSqlStore_GetGroupsByIDs(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestSqlStore_CreateGroup(t *testing.T) {
|
||||
if os.Getenv("CI") == "true" {
|
||||
t.Log("Skipping MySQL test on CI")
|
||||
}
|
||||
t.Setenv("NETBIRD_STORE_ENGINE", string(types.MysqlStoreEngine))
|
||||
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "../testdata/extended-store.sql", t.TempDir())
|
||||
t.Cleanup(cleanup)
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -55,7 +55,6 @@ func (s *SqlStore) ListUsers(ctx context.Context) ([]*types.User, error) {
|
||||
}
|
||||
|
||||
// txDeferFKConstraints defers foreign key constraint checks for the duration of the transaction.
|
||||
// MySQL is already handled by s.transaction (SET FOREIGN_KEY_CHECKS = 0).
|
||||
func (s *SqlStore) txDeferFKConstraints(tx *gorm.DB) error {
|
||||
if s.conn.Engine() == types.SqliteStoreEngine {
|
||||
return tx.Exec("PRAGMA defer_foreign_keys = ON").Error
|
||||
|
||||
@@ -375,7 +375,7 @@ func (s *SqlStore) GetAccountByPeerID(ctx context.Context, peerID string) (*type
|
||||
|
||||
func (s *SqlStore) GetAccountByPeerPubKey(ctx context.Context, peerKey string) (*types.Account, error) {
|
||||
var peer nbpeer.Peer
|
||||
result := s.db.Select("account_id").Take(&peer, GetKeyQueryCondition(s), peerKey)
|
||||
result := s.db.Select("account_id").Take(&peer, keyQueryCondition, peerKey)
|
||||
|
||||
if result.Error != nil {
|
||||
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
|
||||
@@ -394,7 +394,7 @@ func (s *SqlStore) GetAccountByPeerPubKey(ctx context.Context, peerKey string) (
|
||||
func (s *SqlStore) GetAccountIDByPeerPubKey(ctx context.Context, peerKey string) (string, error) {
|
||||
var peer nbpeer.Peer
|
||||
var accountID string
|
||||
result := s.db.Model(&peer).Select("account_id").Where(GetKeyQueryCondition(s), peerKey).Take(&accountID)
|
||||
result := s.db.Model(&peer).Select("account_id").Where(keyQueryCondition, peerKey).Take(&accountID)
|
||||
if result.Error != nil {
|
||||
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
|
||||
return "", status.Errorf(status.NotFound, "account not found: index lookup failed")
|
||||
@@ -483,7 +483,7 @@ func (s *SqlStore) GetPeerByPeerPubKey(ctx context.Context, lockStrength Locking
|
||||
}
|
||||
|
||||
var peer nbpeer.Peer
|
||||
result := tx.Take(&peer, GetKeyQueryCondition(s), peerKey)
|
||||
result := tx.Take(&peer, keyQueryCondition, peerKey)
|
||||
|
||||
if result.Error != nil {
|
||||
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
|
||||
@@ -914,7 +914,7 @@ func (s *SqlStore) GetUserIDByPeerKey(ctx context.Context, lockStrength LockingS
|
||||
var userID string
|
||||
result := tx.Model(&nbpeer.Peer{}).
|
||||
Select("user_id").
|
||||
Take(&userID, GetKeyQueryCondition(s), peerKey)
|
||||
Take(&userID, keyQueryCondition, peerKey)
|
||||
|
||||
if result.Error != nil {
|
||||
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
|
||||
@@ -935,7 +935,7 @@ func (s *SqlStore) GetPeerIDByKey(ctx context.Context, lockStrength LockingStren
|
||||
var peerID string
|
||||
result := tx.Model(&nbpeer.Peer{}).
|
||||
Select("id").
|
||||
Where(GetKeyQueryCondition(s), key).
|
||||
Where(keyQueryCondition, key).
|
||||
Limit(1).
|
||||
Scan(&peerID)
|
||||
if result.Error != nil {
|
||||
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
)
|
||||
|
||||
// TestPgxServiceColumnsMatchGorm guards the Postgres pgx read path against
|
||||
// drifting from the gorm model. The SQLite/MySQL gorm path loads rows by struct,
|
||||
// drifting from the gorm model. The SQLite gorm path loads rows by struct,
|
||||
// so a new column on a model is picked up automatically, but the hand-written
|
||||
// pgx SELECT in sql_store.go must be updated by hand. This test fails when a
|
||||
// gorm column is missing from the pgx column list, which otherwise silently
|
||||
|
||||
@@ -21,8 +21,8 @@ func (s *SqlStore) GetProxyMetrics(ctx context.Context) (ProxyMetrics, error) {
|
||||
var m ProxyMetrics
|
||||
activeCutoff := time.Now().Add(-proxyActiveThreshold)
|
||||
|
||||
// COUNT(DISTINCT ... CASE WHEN ...) is portable across sqlite/postgres
|
||||
// (MySQL too) and keeps the round-trip to one. proxy.StatusConnected
|
||||
// COUNT(DISTINCT ... CASE WHEN ...) is portable across sqlite and postgres
|
||||
// and keeps the round-trip to one. proxy.StatusConnected
|
||||
// is the same string the cluster-capability queries use; the active
|
||||
// window matches the cluster-capability semantics (only proxies
|
||||
// heartbeating within ~2 * heartbeat interval count as connected).
|
||||
@@ -292,7 +292,7 @@ func (s *SqlStore) GetProxyClusters(ctx context.Context, accountID string) ([]pr
|
||||
"cluster_address AS address, "+
|
||||
// COUNT(CASE WHEN ... THEN 1 END) counts only non-NULL — i.e. only
|
||||
// rows that satisfy the predicate — so it works portably across
|
||||
// sqlite/postgres/mysql without dialect-specific FILTER syntax.
|
||||
// sqlite and postgres without dialect-specific FILTER syntax.
|
||||
"COUNT(CASE WHEN status = ? AND last_seen > ? THEN 1 END) AS connected_proxies, "+
|
||||
// MAX(CASE …) > 0 expresses BOOL_OR in a way Postgres tolerates
|
||||
// (Postgres can't MAX a boolean column).
|
||||
|
||||
@@ -18,7 +18,7 @@ import (
|
||||
|
||||
func (s *SqlStore) GetAccountBySetupKey(ctx context.Context, setupKey string) (*types.Account, error) {
|
||||
var key types.SetupKey
|
||||
result := s.db.Select("account_id").Take(&key, GetKeyQueryCondition(s), setupKey)
|
||||
result := s.db.Select("account_id").Take(&key, keyQueryCondition, setupKey)
|
||||
if result.Error != nil {
|
||||
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
|
||||
return nil, status.NewSetupKeyNotFoundError(setupKey)
|
||||
@@ -99,7 +99,7 @@ func (s *SqlStore) getSetupKeys(ctx context.Context, accountID string) ([]types.
|
||||
|
||||
func (s *SqlStore) GetAccountIDBySetupKey(ctx context.Context, setupKey string) (string, error) {
|
||||
var accountID string
|
||||
result := s.db.Model(&types.SetupKey{}).Select("account_id").Where(GetKeyQueryCondition(s), setupKey).Take(&accountID)
|
||||
result := s.db.Model(&types.SetupKey{}).Select("account_id").Where(keyQueryCondition, setupKey).Take(&accountID)
|
||||
if result.Error != nil {
|
||||
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
|
||||
return "", status.NewSetupKeyNotFoundError(setupKey)
|
||||
@@ -123,7 +123,7 @@ func (s *SqlStore) GetSetupKeyBySecret(ctx context.Context, lockStrength Locking
|
||||
|
||||
var setupKey types.SetupKey
|
||||
result := tx.
|
||||
Take(&setupKey, GetKeyQueryCondition(s), key)
|
||||
Take(&setupKey, keyQueryCondition, key)
|
||||
|
||||
if result.Error != nil {
|
||||
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
|
||||
|
||||
@@ -164,7 +164,6 @@ func TestMigrate(t *testing.T) {
|
||||
|
||||
err = migratePreAuto(context.Background(), store.(*SqlStore).db)
|
||||
require.NoError(t, err, "Migration should not fail on migrated db")
|
||||
|
||||
}
|
||||
|
||||
func newAccount(store Store, id int) error {
|
||||
@@ -352,10 +351,6 @@ func TestSqlStore_DatabaseBlocking(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestSqlStore_ExecuteInTransaction_Timeout(t *testing.T) {
|
||||
if os.Getenv("NETBIRD_STORE_ENGINE") == "mysql" {
|
||||
t.Skip("Skipping timeout test for MySQL")
|
||||
}
|
||||
|
||||
t.Setenv("NB_STORE_TRANSACTION_TIMEOUT", "1s")
|
||||
|
||||
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "", t.TempDir())
|
||||
@@ -429,49 +424,6 @@ func TestNewSqliteStore_BusyTimeoutRespectsUserOverride(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSqlStore_ExecuteInTransaction_RestoresForeignKeyChecksOnMysql(t *testing.T) {
|
||||
runTestForAllEngines(t, "", func(t *testing.T, store Store) {
|
||||
sqlStore := store.(*SqlStore)
|
||||
if sqlStore.conn.Engine() != types.MysqlStoreEngine {
|
||||
t.Skip("FOREIGN_KEY_CHECKS is MySQL specific")
|
||||
}
|
||||
sqlDB, err := sqlStore.GetDB().DB()
|
||||
require.NoError(t, err)
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
ctx := context.Background()
|
||||
|
||||
foreignKeyChecks := func() int {
|
||||
var enabled int
|
||||
require.NoError(t, sqlStore.GetDB().Raw("SELECT @@SESSION.foreign_key_checks").Scan(&enabled).Error)
|
||||
return enabled
|
||||
}
|
||||
|
||||
err = store.ExecuteInTransaction(ctx, func(Store) error { return assert.AnError })
|
||||
require.ErrorIs(t, err, assert.AnError)
|
||||
assert.Equal(t, 1, foreignKeyChecks())
|
||||
|
||||
require.Panics(t, func() {
|
||||
_ = store.ExecuteInTransaction(ctx, func(Store) error { panic("boom") })
|
||||
})
|
||||
assert.Equal(t, 1, foreignKeyChecks())
|
||||
|
||||
err = sqlStore.transaction(ctx, func(*gorm.DB) error { return assert.AnError })
|
||||
require.ErrorIs(t, err, assert.AnError)
|
||||
assert.Equal(t, 1, foreignKeyChecks())
|
||||
|
||||
err = store.ExecuteInTransaction(ctx, func(transaction Store) error {
|
||||
bound := transaction.(*SqlStore)
|
||||
require.NoError(t, bound.transaction(ctx, func(*gorm.DB) error { return nil }))
|
||||
var enabled int
|
||||
require.NoError(t, bound.GetDB().Raw("SELECT @@SESSION.foreign_key_checks").Scan(&enabled).Error)
|
||||
assert.Equal(t, 0, enabled, "a savepoint must not re-enable FK checks for the rest of the transaction")
|
||||
return nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, foreignKeyChecks())
|
||||
})
|
||||
}
|
||||
|
||||
func TestSqlStore_Transaction_RollsBackOnError(t *testing.T) {
|
||||
runTestForAllEngines(t, "../testdata/extended-store.sql", func(t *testing.T, store Store) {
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -4,7 +4,6 @@ package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
@@ -20,7 +19,6 @@ import (
|
||||
|
||||
"github.com/google/uuid"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"gorm.io/driver/mysql"
|
||||
"gorm.io/driver/postgres"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/logger"
|
||||
@@ -452,8 +450,6 @@ type AgentNetworkMetrics struct {
|
||||
const (
|
||||
PostgresDsnEnv = "NB_STORE_ENGINE_POSTGRES_DSN"
|
||||
PostgresDsnEnvLegacy = "NETBIRD_STORE_ENGINE_POSTGRES_DSN"
|
||||
mysqlDsnEnv = "NB_STORE_ENGINE_MYSQL_DSN"
|
||||
mysqlDsnEnvLegacy = "NETBIRD_STORE_ENGINE_MYSQL_DSN"
|
||||
)
|
||||
|
||||
// lookupDSNEnv checks the NB_ env var first, then falls back to the legacy NETBIRD_ env var.
|
||||
@@ -464,7 +460,7 @@ func lookupDSNEnv(nbKey, legacyKey string) (string, bool) {
|
||||
return os.LookupEnv(legacyKey)
|
||||
}
|
||||
|
||||
var supportedEngines = []types.Engine{types.SqliteStoreEngine, types.PostgresStoreEngine, types.MysqlStoreEngine}
|
||||
var supportedEngines = []types.Engine{types.SqliteStoreEngine, types.PostgresStoreEngine}
|
||||
|
||||
func getStoreEngineFromEnv() types.Engine {
|
||||
// NETBIRD_STORE_ENGINE supposed to be used in tests. Otherwise, rely on the config file.
|
||||
@@ -474,7 +470,7 @@ func getStoreEngineFromEnv() types.Engine {
|
||||
}
|
||||
|
||||
value := types.Engine(strings.ToLower(kind))
|
||||
if slices.Contains(supportedEngines, value) {
|
||||
if slices.Contains(supportedEngines, value) || value == types.MysqlStoreEngine {
|
||||
return value
|
||||
}
|
||||
|
||||
@@ -540,12 +536,8 @@ func OpenConn(ctx context.Context, kind types.Engine, dataDir string) (*db.Conn,
|
||||
}
|
||||
return db.OpenPostgres(ctx, dsn, db.DefaultPoolConfig)
|
||||
case types.MysqlStoreEngine:
|
||||
log.WithContext(ctx).Info("using MySQL store engine")
|
||||
dsn, ok := lookupDSNEnv(mysqlDsnEnv, mysqlDsnEnvLegacy)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%s is not set", mysqlDsnEnv)
|
||||
}
|
||||
return db.OpenMysql(ctx, dsn)
|
||||
return nil, fmt.Errorf("the MySQL store engine is no longer supported. Please refer to the documentation for " +
|
||||
"migrating to PostgreSQL: https://docs.netbird.io/selfhosted/migration/mysql-to-postgresql")
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported kind of store: %s", kind)
|
||||
}
|
||||
@@ -725,7 +717,7 @@ func NewTestStoreFromSQL(ctx context.Context, filename string, dataDir string) (
|
||||
}
|
||||
|
||||
if filename != "" {
|
||||
err = LoadSQL(conn.DB(nil), filename, translaterForSql(kind))
|
||||
err = LoadSQL(conn.DB(nil), filename)
|
||||
if err != nil {
|
||||
_ = conn.Close()
|
||||
return nil, nil, fmt.Errorf("failed to load SQL file: %v", err)
|
||||
@@ -761,6 +753,7 @@ func NewTestStoreFromSQL(ctx context.Context, filename string, dataDir string) (
|
||||
return nil, nil, fmt.Errorf("failed to create test store after %d attempts: %v", maxRetries, err)
|
||||
}
|
||||
|
||||
// NewTestStore creates a test store and optionally loads SQL after migrating its schema.
|
||||
func NewTestStore(ctx context.Context, filename string, dataDir string) (Store, func(), error) {
|
||||
kind := getStoreEngineFromEnv()
|
||||
if kind == "" {
|
||||
@@ -786,7 +779,7 @@ func NewTestStore(ctx context.Context, filename string, dataDir string) (Store,
|
||||
}
|
||||
|
||||
if filename != "" {
|
||||
err = LoadSQL(sqlStore.db, filename, translaterForSql(kind))
|
||||
err = LoadSQL(sqlStore.db, filename)
|
||||
if err != nil {
|
||||
_ = sqlStore.Close(ctx)
|
||||
return nil, nil, fmt.Errorf("failed to load SQL file: %v", err)
|
||||
@@ -802,18 +795,6 @@ func NewTestStore(ctx context.Context, filename string, dataDir string) (Store,
|
||||
return sqlStore, cleanup, nil
|
||||
}
|
||||
|
||||
// use to translate double-quote escaped reserved names (ansi)
|
||||
// to mysql tick-escaped ones
|
||||
func translaterForSql(kind types.Engine) *strings.Replacer {
|
||||
if kind != types.MysqlStoreEngine {
|
||||
return strings.NewReplacer()
|
||||
}
|
||||
return strings.NewReplacer(
|
||||
"\"key\"", "`key`",
|
||||
"\"groups\"", "`groups`",
|
||||
)
|
||||
}
|
||||
|
||||
func addAllGroupToAccount(ctx context.Context, store Store) error {
|
||||
allAccounts := store.GetAllAccounts(ctx)
|
||||
for _, account := range allAccounts {
|
||||
@@ -843,9 +824,7 @@ func getSqlStoreEngine(ctx context.Context, sqliteStore *SqlStore, kind types.En
|
||||
var err error
|
||||
switch kind {
|
||||
case types.PostgresStoreEngine:
|
||||
store, cleanup, err = newReusedPostgresStore(ctx, sqliteStore, kind)
|
||||
case types.MysqlStoreEngine:
|
||||
store, cleanup, err = newReusedMysqlStore(ctx, sqliteStore, kind)
|
||||
store, cleanup, err = newReusedPostgresStore(ctx, sqliteStore)
|
||||
default:
|
||||
cleanup = func() {
|
||||
// sqlite doesn't need to be cleaned up
|
||||
@@ -877,7 +856,7 @@ func getSqlStoreEngine(ctx context.Context, sqliteStore *SqlStore, kind types.En
|
||||
return store, closeConnection, nil
|
||||
}
|
||||
|
||||
func newReusedPostgresStore(ctx context.Context, sqliteStore *SqlStore, kind types.Engine) (*SqlStore, func(), error) {
|
||||
func newReusedPostgresStore(ctx context.Context, sqliteStore *SqlStore) (*SqlStore, func(), error) {
|
||||
dsn, ok := lookupDSNEnv(PostgresDsnEnv, PostgresDsnEnvLegacy)
|
||||
if !ok || dsn == "" {
|
||||
var err error
|
||||
@@ -891,7 +870,7 @@ func newReusedPostgresStore(ctx context.Context, sqliteStore *SqlStore, kind typ
|
||||
return nil, nil, fmt.Errorf("%s is not set", PostgresDsnEnv)
|
||||
}
|
||||
|
||||
db, err := openDBWithRetry(dsn, kind, 5)
|
||||
db, err := openDBWithRetry(dsn, 5)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to open postgres connection: %v", err)
|
||||
}
|
||||
@@ -902,7 +881,7 @@ func newReusedPostgresStore(ctx context.Context, sqliteStore *SqlStore, kind typ
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
dsn, cleanup, err := createRandomDB(dsn, db, kind, template)
|
||||
dsn, cleanup, err := createRandomDB(dsn, db, template)
|
||||
|
||||
closeGormDB(db)
|
||||
|
||||
@@ -928,74 +907,10 @@ func newReusedPostgresStore(ctx context.Context, sqliteStore *SqlStore, kind typ
|
||||
return store, cleanup, nil
|
||||
}
|
||||
|
||||
func newReusedMysqlStore(ctx context.Context, sqliteStore *SqlStore, kind types.Engine) (*SqlStore, func(), error) {
|
||||
dsn, ok := lookupDSNEnv(mysqlDsnEnv, mysqlDsnEnvLegacy)
|
||||
if !ok || dsn == "" {
|
||||
var err error
|
||||
_, dsn, err = testutil.CreateMysqlTestContainer()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
if dsn == "" {
|
||||
return nil, nil, fmt.Errorf("%s is not set", mysqlDsnEnv)
|
||||
}
|
||||
|
||||
db, err := openDBWithRetry(dsn, kind, 5)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to open mysql connection: %v", err)
|
||||
}
|
||||
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to get underlying sql.DB: %v", err)
|
||||
}
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
sqlDB.SetMaxIdleConns(1)
|
||||
|
||||
tableDDL, err := mysqlSchemaTemplate(ctx, dsn, db)
|
||||
if err != nil {
|
||||
sqlDB.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
dsn, cleanup, err := createRandomDB(dsn, db, kind, "")
|
||||
|
||||
sqlDB.Close()
|
||||
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
if err := cloneMysqlSchema(ctx, dsn, tableDDL); err != nil {
|
||||
cleanup()
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
var store *SqlStore
|
||||
if sqliteStore != nil {
|
||||
store, err = newMysqlStoreFromSqlStore(ctx, sqliteStore, dsn, nil, true)
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, err
|
||||
}
|
||||
} else {
|
||||
store, err = NewMysqlStore(ctx, dsn, nil, false)
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return store, cleanup, nil
|
||||
}
|
||||
|
||||
// schemaTemplates remembers, per engine and server, a database that went
|
||||
// through the full migration once in this process. Every later test database
|
||||
// is cloned from it, so a test pays for CREATE DATABASE and a schema copy
|
||||
// instead of the 40-table AutoMigrate plus every pre and post migration, which
|
||||
// is what made each MySQL test store cost well over a second in CI.
|
||||
// instead of the 40-table AutoMigrate plus every pre and post migration.
|
||||
var (
|
||||
schemaTemplatesMu sync.Mutex
|
||||
schemaTemplates = map[string]*schemaTemplate{}
|
||||
@@ -1003,10 +918,6 @@ var (
|
||||
|
||||
type schemaTemplate struct {
|
||||
dbName string
|
||||
// tableDDL holds the CREATE TABLE statements of the template. MySQL has no
|
||||
// server-side database template, so the schema is replayed statement by
|
||||
// statement into each test database.
|
||||
tableDDL []string
|
||||
}
|
||||
|
||||
func schemaTemplateKey(engine types.Engine, dsn string) string {
|
||||
@@ -1046,112 +957,6 @@ func postgresSchemaTemplate(ctx context.Context, baseDSN string, admin *gorm.DB)
|
||||
return name, nil
|
||||
}
|
||||
|
||||
// mysqlSchemaTemplate returns the CREATE TABLE statements of a fully migrated
|
||||
// database, migrating one on first use.
|
||||
func mysqlSchemaTemplate(ctx context.Context, baseDSN string, admin *gorm.DB) ([]string, error) {
|
||||
schemaTemplatesMu.Lock()
|
||||
defer schemaTemplatesMu.Unlock()
|
||||
|
||||
key := schemaTemplateKey(types.MysqlStoreEngine, baseDSN)
|
||||
if tpl, ok := schemaTemplates[key]; ok {
|
||||
return tpl.tableDDL, nil
|
||||
}
|
||||
|
||||
name := newTestDBName("test_template")
|
||||
if err := admin.Exec(fmt.Sprintf("CREATE DATABASE %s", name)).Error; err != nil {
|
||||
return nil, fmt.Errorf("create mysql template database: %w", err)
|
||||
}
|
||||
|
||||
tplStore, err := NewMysqlStore(ctx, replaceDBName(baseDSN, name), nil, false)
|
||||
if err != nil {
|
||||
dropDatabase(admin, name)
|
||||
return nil, fmt.Errorf("migrate mysql template database: %w", err)
|
||||
}
|
||||
tableDDL, err := mysqlTableDDL(ctx, tplStore.db, name)
|
||||
tplStore.Close(ctx)
|
||||
if err != nil {
|
||||
dropDatabase(admin, name)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
schemaTemplates[key] = &schemaTemplate{dbName: name, tableDDL: tableDDL}
|
||||
return tableDDL, nil
|
||||
}
|
||||
|
||||
func mysqlTableDDL(ctx context.Context, db *gorm.DB, dbName string) ([]string, error) {
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tables, err := mysqlTableNames(ctx, sqlDB, dbName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tableDDL := make([]string, 0, len(tables))
|
||||
for _, table := range tables {
|
||||
var name, createStmt string
|
||||
row := sqlDB.QueryRowContext(ctx, fmt.Sprintf("SHOW CREATE TABLE %s.%s", dbName, table))
|
||||
if err := row.Scan(&name, &createStmt); err != nil {
|
||||
return nil, fmt.Errorf("read create statement of %s: %w", table, err)
|
||||
}
|
||||
tableDDL = append(tableDDL, createStmt)
|
||||
}
|
||||
return tableDDL, nil
|
||||
}
|
||||
|
||||
func mysqlTableNames(ctx context.Context, sqlDB *sql.DB, dbName string) ([]string, error) {
|
||||
rows, err := sqlDB.QueryContext(ctx, fmt.Sprintf("SHOW TABLES FROM %s", dbName))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list template tables: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var tables []string
|
||||
for rows.Next() {
|
||||
var table string
|
||||
if err := rows.Scan(&table); err != nil {
|
||||
return nil, fmt.Errorf("scan template table name: %w", err)
|
||||
}
|
||||
tables = append(tables, table)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("list template tables: %w", err)
|
||||
}
|
||||
return tables, nil
|
||||
}
|
||||
|
||||
// cloneMysqlSchema replays the template's CREATE TABLE statements into the
|
||||
// database the DSN points at.
|
||||
func cloneMysqlSchema(ctx context.Context, dsn string, tableDDL []string) error {
|
||||
gormDB, err := gorm.Open(mysql.Open(db.MysqlDSN(dsn)), db.GormConfig())
|
||||
if err != nil {
|
||||
return fmt.Errorf("connect to test database: %w", err)
|
||||
}
|
||||
sqlDB, err := gormDB.DB()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer sqlDB.Close()
|
||||
|
||||
// The statements come out of SHOW TABLES in name order, not dependency
|
||||
// order, and their foreign keys reference tables of the session's default
|
||||
// database. Pin a single connection so the session setting below covers
|
||||
// every statement, and connect straight to the new database so unqualified
|
||||
// references land there.
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
if _, err := sqlDB.ExecContext(ctx, "SET FOREIGN_KEY_CHECKS = 0"); err != nil {
|
||||
return fmt.Errorf("disable foreign key checks: %w", err)
|
||||
}
|
||||
for _, stmt := range tableDDL {
|
||||
if _, err := sqlDB.ExecContext(ctx, stmt); err != nil {
|
||||
return fmt.Errorf("replay table definition: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// dropDatabase removes a template that never became usable, so a failed setup
|
||||
// does not leave it behind on a shared server. The server may still be tearing
|
||||
// down the sessions the failed migration held, so the drop retries while
|
||||
@@ -1168,18 +973,12 @@ func closeGormDB(db *gorm.DB) {
|
||||
}
|
||||
}
|
||||
|
||||
func openDBWithRetry(dsn string, engine types.Engine, maxRetries int) (*gorm.DB, error) {
|
||||
func openDBWithRetry(dsn string, maxRetries int) (*gorm.DB, error) {
|
||||
var gormDB *gorm.DB
|
||||
var err error
|
||||
|
||||
for i := range maxRetries {
|
||||
switch engine {
|
||||
case types.PostgresStoreEngine:
|
||||
gormDB, err = gorm.Open(postgres.Open(dsn), &gorm.Config{})
|
||||
case types.MysqlStoreEngine:
|
||||
gormDB, err = gorm.Open(mysql.Open(db.MysqlDSN(dsn)), &gorm.Config{})
|
||||
}
|
||||
|
||||
gormDB, err = gorm.Open(postgres.Open(dsn), &gorm.Config{})
|
||||
if err == nil {
|
||||
return gormDB, nil
|
||||
}
|
||||
@@ -1193,13 +992,13 @@ func openDBWithRetry(dsn string, engine types.Engine, maxRetries int) (*gorm.DB,
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// createRandomDB creates a uniquely named database for one test. On postgres a
|
||||
// non-empty template is copied server-side with CREATE DATABASE ... TEMPLATE.
|
||||
func createRandomDB(dsn string, admin *gorm.DB, engine types.Engine, template string) (string, func(), error) {
|
||||
// createRandomDB creates a uniquely named database for one test. A non-empty
|
||||
// template is copied server-side with CREATE DATABASE ... TEMPLATE.
|
||||
func createRandomDB(dsn string, admin *gorm.DB, template string) (string, func(), error) {
|
||||
dbName := newTestDBName("test_db")
|
||||
|
||||
createStmt := fmt.Sprintf("CREATE DATABASE %s", dbName)
|
||||
if template != "" && engine == types.PostgresStoreEngine {
|
||||
if template != "" {
|
||||
createStmt = fmt.Sprintf("CREATE DATABASE %s TEMPLATE %s", dbName, template)
|
||||
}
|
||||
if err := execWithTemplateRetry(admin, createStmt); err != nil {
|
||||
@@ -1209,59 +1008,28 @@ func createRandomDB(dsn string, admin *gorm.DB, engine types.Engine, template st
|
||||
originalDSN := dsn
|
||||
|
||||
cleanup := func() {
|
||||
var dropDB *gorm.DB
|
||||
var err error
|
||||
|
||||
switch engine {
|
||||
case types.PostgresStoreEngine:
|
||||
dropDB, err = gorm.Open(postgres.Open(originalDSN), &gorm.Config{
|
||||
SkipDefaultTransaction: true,
|
||||
PrepareStmt: false,
|
||||
Logger: logger.Default.LogMode(logger.Info),
|
||||
})
|
||||
if err != nil {
|
||||
log.Errorf("failed to connect for dropping database %s: %v", dbName, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if sqlDB, _ := dropDB.DB(); sqlDB != nil {
|
||||
sqlDB.Close()
|
||||
}
|
||||
}()
|
||||
|
||||
dropDB, err := gorm.Open(postgres.Open(originalDSN), &gorm.Config{
|
||||
SkipDefaultTransaction: true,
|
||||
PrepareStmt: false,
|
||||
Logger: logger.Default.LogMode(logger.Info),
|
||||
})
|
||||
if err != nil {
|
||||
log.Errorf("failed to connect for dropping database %s: %v", dbName, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if sqlDB, _ := dropDB.DB(); sqlDB != nil {
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
sqlDB.SetMaxIdleConns(0)
|
||||
sqlDB.SetConnMaxLifetime(time.Second)
|
||||
sqlDB.Close()
|
||||
}
|
||||
}()
|
||||
|
||||
err = dropDB.Exec(fmt.Sprintf("DROP DATABASE IF EXISTS %s WITH (FORCE)", dbName)).Error
|
||||
|
||||
case types.MysqlStoreEngine:
|
||||
dropDB, err = gorm.Open(mysql.Open(db.MysqlDSN(originalDSN)), &gorm.Config{
|
||||
SkipDefaultTransaction: true,
|
||||
PrepareStmt: false,
|
||||
})
|
||||
if err != nil {
|
||||
log.Errorf("failed to connect for dropping database %s: %v", dbName, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if sqlDB, _ := dropDB.DB(); sqlDB != nil {
|
||||
sqlDB.Close()
|
||||
}
|
||||
}()
|
||||
|
||||
if sqlDB, _ := dropDB.DB(); sqlDB != nil {
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
sqlDB.SetMaxIdleConns(0)
|
||||
sqlDB.SetConnMaxLifetime(time.Second)
|
||||
}
|
||||
|
||||
err = dropDB.Exec(fmt.Sprintf("DROP DATABASE IF EXISTS %s", dbName)).Error
|
||||
if sqlDB, _ := dropDB.DB(); sqlDB != nil {
|
||||
sqlDB.SetMaxOpenConns(1)
|
||||
sqlDB.SetMaxIdleConns(0)
|
||||
sqlDB.SetConnMaxLifetime(time.Second)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
if err := dropDB.Exec(fmt.Sprintf("DROP DATABASE IF EXISTS %s WITH (FORCE)", dbName)).Error; err != nil {
|
||||
log.Errorf("failed to drop database %s: %v", dbName, err)
|
||||
}
|
||||
}
|
||||
@@ -1288,7 +1056,8 @@ func replaceDBName(dsn, newDBName string) string {
|
||||
return re.ReplaceAllString(dsn, `${pre}`+newDBName+`${post}`)
|
||||
}
|
||||
|
||||
func LoadSQL(db *gorm.DB, filepath string, translator *strings.Replacer) error {
|
||||
// LoadSQL executes the SQL statements from the given file.
|
||||
func LoadSQL(db *gorm.DB, filepath string) error {
|
||||
sqlContent, err := os.ReadFile(filepath)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -1297,7 +1066,7 @@ func LoadSQL(db *gorm.DB, filepath string, translator *strings.Replacer) error {
|
||||
queries := strings.Split(string(sqlContent), ";")
|
||||
|
||||
for _, query := range queries {
|
||||
query = translator.Replace(strings.TrimSpace(query))
|
||||
query = strings.TrimSpace(query)
|
||||
if query != "" {
|
||||
err := db.Exec(query).Error
|
||||
if err != nil {
|
||||
|
||||
@@ -8,74 +8,12 @@ import (
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
"github.com/testcontainers/testcontainers-go/modules/mysql"
|
||||
"github.com/testcontainers/testcontainers-go/modules/postgres"
|
||||
testcontainersredis "github.com/testcontainers/testcontainers-go/modules/redis"
|
||||
"github.com/testcontainers/testcontainers-go/wait"
|
||||
)
|
||||
|
||||
var (
|
||||
pgContainer *postgres.PostgresContainer
|
||||
mysqlContainer *mysql.MySQLContainer
|
||||
)
|
||||
|
||||
// CreateMysqlTestContainer creates a new MySQL container for testing.
|
||||
func CreateMysqlTestContainer() (func(), string, error) {
|
||||
ctx := context.Background()
|
||||
|
||||
if mysqlContainer != nil {
|
||||
connStr, err := mysqlContainer.ConnectionString(ctx)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
return noOpCleanup, connStr, nil
|
||||
}
|
||||
|
||||
var err error
|
||||
mysqlContainer, err = mysql.Run(ctx,
|
||||
"mlsmaycon/warmed-mysql:8",
|
||||
mysql.WithDatabase("testing"),
|
||||
mysql.WithUsername("root"),
|
||||
mysql.WithPassword("testing"),
|
||||
// Every test creates and drops a database with about 40 tables, so with
|
||||
// the server defaults the run is dominated by durability work: each
|
||||
// CREATE TABLE fsyncs the redo log, the binary log and the doublewrite
|
||||
// buffer. None of it protects anything in a container that is discarded
|
||||
// after the run. Tables stay in per-table files on purpose: in the shared
|
||||
// system tablespace the cost of every CREATE and DROP grew with the number
|
||||
// of databases the run had already created.
|
||||
testcontainers.WithCmd("mysqld",
|
||||
"--innodb-flush-log-at-trx-commit=0",
|
||||
"--innodb-doublewrite=OFF",
|
||||
"--skip-log-bin",
|
||||
),
|
||||
testcontainers.WithWaitStrategy(
|
||||
wait.ForLog("/usr/sbin/mysqld: ready for connections").
|
||||
WithOccurrence(1).WithStartupTimeout(15*time.Second).WithPollInterval(100*time.Millisecond),
|
||||
),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
if mysqlContainer != nil {
|
||||
timeoutCtx, cancelFunc := context.WithTimeout(ctx, 1*time.Second)
|
||||
defer cancelFunc()
|
||||
if err = mysqlContainer.Terminate(timeoutCtx); err != nil {
|
||||
log.WithContext(ctx).Warnf("failed to stop mysql container %s: %s", mysqlContainer.GetContainerID(), err)
|
||||
}
|
||||
mysqlContainer = nil // reset the container to allow recreation
|
||||
}
|
||||
}
|
||||
|
||||
talksConn, err := mysqlContainer.ConnectionString(ctx)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
return cleanup, talksConn, nil
|
||||
}
|
||||
var pgContainer *postgres.PostgresContainer
|
||||
|
||||
// CreatePostgresTestContainer creates a new PostgreSQL container for testing.
|
||||
func CreatePostgresTestContainer() (func(), string, error) {
|
||||
|
||||
@@ -8,12 +8,6 @@ func CreatePostgresTestContainer() (func(), string, error) {
|
||||
}, "", nil
|
||||
}
|
||||
|
||||
func CreateMysqlTestContainer() (func(), string, error) {
|
||||
return func() {
|
||||
// Empty function for MySQL
|
||||
}, "", nil
|
||||
}
|
||||
|
||||
func CreateRedisTestContainer() (func(), string, error) {
|
||||
return func() {
|
||||
// Empty function for Redis
|
||||
|
||||
@@ -8,5 +8,5 @@ const (
|
||||
PostgresStoreEngine = db.PostgresStoreEngine
|
||||
FileStoreEngine Engine = "jsonfile"
|
||||
SqliteStoreEngine = db.SqliteStoreEngine
|
||||
MysqlStoreEngine = db.MysqlStoreEngine
|
||||
MysqlStoreEngine Engine = "mysql"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user