Add migration support for mysql

cmd/cc-backend/main.go

@@ -114,7 +114,7 @@ func main() {
 	}
 
 	if flagMigrateDB {
-		repository.MigrateDB(config.Keys.DB)
+		repository.MigrateDB(config.Keys.DBDriver, config.Keys.DB)
 		os.Exit(0)
 	}
 

internal/repository/dbConnection.go

@@ -55,7 +55,7 @@ func Connect(driver string, db string) {
 		}
 
 		dbConnInstance = &DBConnection{DB: dbHandle}
-		checkDBVersion(dbHandle.DB)
+		checkDBVersion(driver, dbHandle.DB)
 	})
 }
 

internal/repository/migration.go

@@ -12,29 +12,49 @@ import (
 
 	"github.com/ClusterCockpit/cc-backend/pkg/log"
 	"github.com/golang-migrate/migrate/v4"
+	"github.com/golang-migrate/migrate/v4/database/mysql"
 	"github.com/golang-migrate/migrate/v4/database/sqlite3"
 	"github.com/golang-migrate/migrate/v4/source/iofs"
 )
 
 const supportedVersion uint = 2
 
-//go:embed migrations/*.sql
+//go:embed migrations/*
 var migrationFiles embed.FS
 
-func checkDBVersion(db *sql.DB) {
+func checkDBVersion(backend string, db *sql.DB) {
+	var m *migrate.Migrate
+
+	if backend == "sqlite3" {
+
 		driver, err := sqlite3.WithInstance(db, &sqlite3.Config{})
 		if err != nil {
 			log.Fatal(err)
 		}
-	d, err := iofs.New(migrationFiles, "migrations")
+		d, err := iofs.New(migrationFiles, "migrations/sqlite3")
 		if err != nil {
 			log.Fatal(err)
 		}
 
-	m, err := migrate.NewWithInstance("iofs", d, "sqlite3", driver)
+		m, err = migrate.NewWithInstance("iofs", d, "sqlite3", driver)
 		if err != nil {
 			log.Fatal(err)
 		}
+	} else if backend == "mysql" {
+		driver, err := mysql.WithInstance(db, &mysql.Config{})
+		if err != nil {
+			log.Fatal(err)
+		}
+		d, err := iofs.New(migrationFiles, "migrations/mysql")
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		m, err = migrate.NewWithInstance("iofs", d, "mysql", driver)
+		if err != nil {
+			log.Fatal(err)
+		}
+	}
 
 	v, _, err := m.Version()
 	if err != nil {
@@ -56,16 +76,31 @@ func checkDBVersion(db *sql.DB) {
 	}
 }
 
-func MigrateDB(db string) {
+func MigrateDB(backend string, db string) {
-	d, err := iofs.New(migrationFiles, "migrations")
+	var m *migrate.Migrate
+
+	if backend == "sqlite3" {
+		d, err := iofs.New(migrationFiles, "migrations/sqlite3")
 		if err != nil {
 			log.Fatal(err)
 		}
 
-	m, err := migrate.NewWithSourceInstance("iofs", d, fmt.Sprintf("sqlite3://%s?_foreign_keys=on", db))
+		m, err = migrate.NewWithSourceInstance("iofs", d, fmt.Sprintf("sqlite3://%s?_foreign_keys=on", db))
 		if err != nil {
 			log.Fatal(err)
 		}
+	} else if backend == "mysql" {
+		d, err := iofs.New(migrationFiles, "migrations/mysql")
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		m, err = migrate.NewWithSourceInstance("iofs", d, fmt.Sprintf("mysql://%s?multiStatements=true", db))
+		if err != nil {
+			log.Fatal(err)
+		}
+	}
+
 	if err := m.Up(); err != nil {
 		log.Fatal(err)
 	}

internal/repository/migrations/mysql/01_init-schema.up.sql

@@ -0,0 +1,62 @@
+CREATE TABLE IF NOT EXISTS job (
+    id                INTEGER AUTO_INCREMENT PRIMARY KEY ,
+    job_id            BIGINT NOT NULL,
+    cluster           VARCHAR(255) NOT NULL,
+    subcluster        VARCHAR(255) NOT NULL,
+    start_time        BIGINT NOT NULL, -- Unix timestamp
+
+    user              VARCHAR(255) NOT NULL,
+    project           VARCHAR(255) NOT NULL,
+    `partition`       VARCHAR(255) NOT NULL,
+    array_job_id      BIGINT NOT NULL,
+    duration          INT NOT NULL DEFAULT 0,
+    walltime          INT NOT NULL DEFAULT 0,
+    job_state         VARCHAR(255) NOT NULL 
+    CHECK(job_state IN ('running', 'completed', 'failed', 'cancelled',
+            'stopped', 'timeout', 'preempted', 'out_of_memory')),
+    meta_data         TEXT,          -- JSON
+    resources         TEXT NOT NULL, -- JSON
+
+    num_nodes         INT NOT NULL,
+    num_hwthreads     INT NOT NULL,
+    num_acc           INT NOT NULL,
+    smt               TINYINT NOT NULL DEFAULT 1 CHECK(smt               IN (0, 1   )),
+    exclusive         TINYINT NOT NULL DEFAULT 1 CHECK(exclusive         IN (0, 1, 2)),
+    monitoring_status TINYINT NOT NULL DEFAULT 1 CHECK(monitoring_status IN (0, 1, 2, 3)),
+
+    mem_used_max        REAL NOT NULL DEFAULT 0.0,
+    flops_any_avg       REAL NOT NULL DEFAULT 0.0,
+    mem_bw_avg          REAL NOT NULL DEFAULT 0.0,
+    load_avg            REAL NOT NULL DEFAULT 0.0,
+    net_bw_avg          REAL NOT NULL DEFAULT 0.0,
+    net_data_vol_total  REAL NOT NULL DEFAULT 0.0,
+    file_bw_avg         REAL NOT NULL DEFAULT 0.0,
+    file_data_vol_total REAL NOT NULL DEFAULT 0.0);
+
+CREATE TABLE IF NOT EXISTS tag (
+    id       INTEGER PRIMARY KEY,
+    tag_type VARCHAR(255) NOT NULL,
+    tag_name VARCHAR(255) NOT NULL,
+    CONSTRAINT be_unique UNIQUE (tag_type, tag_name));
+
+CREATE TABLE IF NOT EXISTS jobtag (
+    job_id INTEGER,
+    tag_id INTEGER,
+    PRIMARY KEY (job_id, tag_id),
+    FOREIGN KEY (job_id) REFERENCES job (id) ON DELETE CASCADE,
+    FOREIGN KEY (tag_id) REFERENCES tag (id) ON DELETE CASCADE);
+
+CREATE TABLE IF NOT EXISTS configuration (
+	username varchar(255),
+	confkey  varchar(255),
+	value    varchar(255),
+	PRIMARY KEY (username, confkey),
+	FOREIGN KEY (username) REFERENCES user (username) ON DELETE CASCADE ON UPDATE NO ACTION);
+
+CREATE TABLE IF NOT EXISTS user (
+	username varchar(255) PRIMARY KEY NOT NULL,
+	password varchar(255) DEFAULT NULL,
+	ldap     tinyint      NOT NULL DEFAULT 0, /* col called "ldap" for historic reasons, fills the "AuthSource" */
+	name     varchar(255) DEFAULT NULL,
+	roles    varchar(255) NOT NULL DEFAULT "[]",
+	email    varchar(255) DEFAULT NULL);

internal/repository/migrations/sqlite3/01_init-schema.down.sql

@@ -0,0 +1,5 @@
+DROP TABLE IF EXISTS job;
+DROP TABLE IF EXISTS tags;
+DROP TABLE IF EXISTS jobtag;
+DROP TABLE IF EXISTS configuration;
+DROP TABLE IF EXISTS user;

internal/repository/migrations/sqlite3/02_add-index.down.sql

@@ -0,0 +1,5 @@
+DROP INDEX IF EXISTS job_stats;
+DROP INDEX IF EXISTS job_by_user;
+DROP INDEX IF EXISTS job_by_starttime;
+DROP INDEX IF EXISTS job_by_job_id;
+DROP INDEX IF EXISTS job_by_state;

internal/repository/migrations/sqlite3/02_add-index.up.sql

@@ -0,0 +1,5 @@
+CREATE INDEX IF NOT EXISTS job_stats        ON job (cluster,subcluster,user);
+CREATE INDEX IF NOT EXISTS job_by_user      ON job (user);
+CREATE INDEX IF NOT EXISTS job_by_starttime ON job (start_time);
+CREATE INDEX IF NOT EXISTS job_by_job_id    ON job (job_id);
+CREATE INDEX IF NOT EXISTS job_by_state     ON job (job_state);

test/integration_test.go

@@ -269,7 +269,7 @@ func setup(t *testing.T) *api.RestApi {
 		t.Fatal(err)
 	}
 	dbfilepath := filepath.Join(tmpdir, "test.db")
-	repository.MigrateDB(dbfilepath)
+	repository.MigrateDB("sqlite3", dbfilepath)
 
 	cfgFilePath := filepath.Join(tmpdir, "config.json")
 	if err := os.WriteFile(cfgFilePath, []byte(testconfig), 0666); err != nil {