Refactor directory structure

internal/repository/dbConnection.go

@@ -0,0 +1,58 @@
+package repository
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+)
+
+var (
+	dbConnOnce     sync.Once
+	dbConnInstance *DBConnection
+)
+
+type DBConnection struct {
+	DB *sqlx.DB
+}
+
+func Connect(driver string, db string) {
+	var err error
+	var dbHandle *sqlx.DB
+
+	dbConnOnce.Do(func() {
+		if driver == "sqlite3" {
+			dbHandle, err = sqlx.Open("sqlite3", fmt.Sprintf("%s?_foreign_keys=on", db))
+			if err != nil {
+				log.Fatal(err)
+			}
+
+			// sqlite does not multithread. Having more than one connection open would just mean
+			// waiting for locks.
+			dbHandle.SetMaxOpenConns(1)
+		} else if driver == "mysql" {
+			dbHandle, err = sqlx.Open("mysql", fmt.Sprintf("%s?multiStatements=true", db))
+			if err != nil {
+				log.Fatal(err)
+			}
+
+			dbHandle.SetConnMaxLifetime(time.Minute * 3)
+			dbHandle.SetMaxOpenConns(10)
+			dbHandle.SetMaxIdleConns(10)
+		} else {
+			log.Fatalf("unsupported database driver: %s", driver)
+		}
+
+		dbConnInstance = &DBConnection{DB: dbHandle}
+	})
+}
+
+func GetConnection() *DBConnection {
+	if dbConnInstance == nil {
+		log.Fatalf("Database connection not initialized!")
+	}
+
+	return dbConnInstance
+}

internal/repository/import.go

@@ -0,0 +1,155 @@
+package repository
+
+import (
+	"bytes"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/ClusterCockpit/cc-backend/internal/config"
+	"github.com/ClusterCockpit/cc-backend/internal/metricdata"
+	"github.com/ClusterCockpit/cc-backend/pkg/log"
+	"github.com/ClusterCockpit/cc-backend/pkg/schema"
+)
+
+const NamedJobInsert string = `INSERT INTO job (
+	job_id, user, project, cluster, subcluster, ` + "`partition`" + `, array_job_id, num_nodes, num_hwthreads, num_acc,
+	exclusive, monitoring_status, smt, job_state, start_time, duration, walltime, resources, meta_data,
+	mem_used_max, flops_any_avg, mem_bw_avg, load_avg, net_bw_avg, net_data_vol_total, file_bw_avg, file_data_vol_total
+) VALUES (
+	:job_id, :user, :project, :cluster, :subcluster, :partition, :array_job_id, :num_nodes, :num_hwthreads, :num_acc,
+	:exclusive, :monitoring_status, :smt, :job_state, :start_time, :duration, :walltime, :resources, :meta_data,
+	:mem_used_max, :flops_any_avg, :mem_bw_avg, :load_avg, :net_bw_avg, :net_data_vol_total, :file_bw_avg, :file_data_vol_total
+);`
+
+// Import all jobs specified as `<path-to-meta.json>:<path-to-data.json>,...`
+func (r *JobRepository) HandleImportFlag(flag string) error {
+	for _, pair := range strings.Split(flag, ",") {
+		files := strings.Split(pair, ":")
+		if len(files) != 2 {
+			return fmt.Errorf("invalid import flag format")
+		}
+
+		raw, err := os.ReadFile(files[0])
+		if err != nil {
+			return err
+		}
+
+		dec := json.NewDecoder(bytes.NewReader(raw))
+		dec.DisallowUnknownFields()
+		jobMeta := schema.JobMeta{BaseJob: schema.JobDefaults}
+		if err := dec.Decode(&jobMeta); err != nil {
+			return err
+		}
+
+		raw, err = os.ReadFile(files[1])
+		if err != nil {
+			return err
+		}
+
+		dec = json.NewDecoder(bytes.NewReader(raw))
+		dec.DisallowUnknownFields()
+		jobData := schema.JobData{}
+		if err := dec.Decode(&jobData); err != nil {
+			return err
+		}
+
+		if err := r.ImportJob(&jobMeta, &jobData); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *JobRepository) ImportJob(jobMeta *schema.JobMeta, jobData *schema.JobData) (err error) {
+	jobMeta.MonitoringStatus = schema.MonitoringStatusArchivingSuccessful
+	if err := metricdata.ImportJob(jobMeta, jobData); err != nil {
+		return err
+	}
+
+	if job, err := r.Find(&jobMeta.JobID, &jobMeta.Cluster, &jobMeta.StartTime); err != sql.ErrNoRows {
+		if err != nil {
+			return err
+		}
+
+		return fmt.Errorf("a job with that jobId, cluster and startTime does already exist (dbid: %d)", job.ID)
+	}
+
+	job := schema.Job{
+		BaseJob:       jobMeta.BaseJob,
+		StartTime:     time.Unix(jobMeta.StartTime, 0),
+		StartTimeUnix: jobMeta.StartTime,
+	}
+
+	// TODO: Other metrics...
+	job.FlopsAnyAvg = loadJobStat(jobMeta, "flops_any")
+	job.MemBwAvg = loadJobStat(jobMeta, "mem_bw")
+	job.NetBwAvg = loadJobStat(jobMeta, "net_bw")
+	job.FileBwAvg = loadJobStat(jobMeta, "file_bw")
+	job.RawResources, err = json.Marshal(job.Resources)
+	if err != nil {
+		return err
+	}
+	job.RawMetaData, err = json.Marshal(job.MetaData)
+	if err != nil {
+		return err
+	}
+
+	if err := SanityChecks(&job.BaseJob); err != nil {
+		return err
+	}
+
+	res, err := r.DB.NamedExec(NamedJobInsert, job)
+	if err != nil {
+		return err
+	}
+
+	id, err := res.LastInsertId()
+	if err != nil {
+		return err
+	}
+
+	for _, tag := range job.Tags {
+		if _, err := r.AddTagOrCreate(id, tag.Type, tag.Name); err != nil {
+			return err
+		}
+	}
+
+	log.Infof("Successfully imported a new job (jobId: %d, cluster: %s, dbid: %d)", job.JobID, job.Cluster, id)
+	return nil
+}
+
+// This function also sets the subcluster if necessary!
+func SanityChecks(job *schema.BaseJob) error {
+	if c := config.GetCluster(job.Cluster); c == nil {
+		return fmt.Errorf("no such cluster: %#v", job.Cluster)
+	}
+	if err := config.AssignSubCluster(job); err != nil {
+		return err
+	}
+	if !job.State.Valid() {
+		return fmt.Errorf("not a valid job state: %#v", job.State)
+	}
+	if len(job.Resources) == 0 || len(job.User) == 0 {
+		return fmt.Errorf("'resources' and 'user' should not be empty")
+	}
+	if job.NumAcc < 0 || job.NumHWThreads < 0 || job.NumNodes < 1 {
+		return fmt.Errorf("'numNodes', 'numAcc' or 'numHWThreads' invalid")
+	}
+	if len(job.Resources) != int(job.NumNodes) {
+		return fmt.Errorf("len(resources) does not equal numNodes (%d vs %d)", len(job.Resources), job.NumNodes)
+	}
+
+	return nil
+}
+
+func loadJobStat(job *schema.JobMeta, metric string) float64 {
+	if stats, ok := job.Statistics[metric]; ok {
+		return stats.Avg
+	}
+
+	return 0.0
+}

internal/repository/init.go

@@ -0,0 +1,271 @@
+package repository
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/ClusterCockpit/cc-backend/pkg/log"
+	"github.com/ClusterCockpit/cc-backend/pkg/schema"
+	"github.com/jmoiron/sqlx"
+)
+
+// `AUTO_INCREMENT` is in a comment because of this hack:
+// https://stackoverflow.com/a/41028314 (sqlite creates unique ids automatically)
+const JobsDBSchema string = `
+	DROP TABLE IF EXISTS jobtag;
+	DROP TABLE IF EXISTS job;
+	DROP TABLE IF EXISTS tag;
+
+	CREATE TABLE job (
+		id                INTEGER PRIMARY KEY /*!40101 AUTO_INCREMENT */,
+		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, -- partition is a keyword in mysql -.-
+		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 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 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);
+`
+
+// Indexes are created after the job-archive is traversed for faster inserts.
+const JobsDbIndexes string = `
+	CREATE INDEX job_by_user      ON job (user);
+	CREATE INDEX job_by_starttime ON job (start_time);
+	CREATE INDEX job_by_job_id    ON job (job_id);
+	CREATE INDEX job_by_state     ON job (job_state);
+`
+
+// Delete the tables "job", "tag" and "jobtag" from the database and
+// repopulate them using the jobs found in `archive`.
+func InitDB(db *sqlx.DB, archive string) error {
+	starttime := time.Now()
+	log.Print("Building job table...")
+
+	// Basic database structure:
+	_, err := db.Exec(JobsDBSchema)
+	if err != nil {
+		return err
+	}
+
+	clustersDir, err := os.ReadDir(archive)
+	if err != nil {
+		return err
+	}
+
+	if err != nil {
+		return err
+	}
+
+	// Inserts are bundled into transactions because in sqlite,
+	// that speeds up inserts A LOT.
+	tx, err := db.Beginx()
+	if err != nil {
+		return err
+	}
+
+	stmt, err := tx.PrepareNamed(NamedJobInsert)
+	if err != nil {
+		return err
+	}
+
+	// Not using log.Print because we want the line to end with `\r` and
+	// this function is only ever called when a special command line flag
+	// is passed anyways.
+	fmt.Printf("%d jobs inserted...\r", 0)
+	i := 0
+	tags := make(map[string]int64)
+	handleDirectory := func(filename string) error {
+		// Bundle 100 inserts into one transaction for better performance:
+		if i%100 == 0 {
+			if tx != nil {
+				if err := tx.Commit(); err != nil {
+					return err
+				}
+			}
+
+			tx, err = db.Beginx()
+			if err != nil {
+				return err
+			}
+
+			stmt = tx.NamedStmt(stmt)
+			fmt.Printf("%d jobs inserted...\r", i)
+		}
+
+		err := loadJob(tx, stmt, tags, filename)
+		if err == nil {
+			i += 1
+		}
+
+		return err
+	}
+
+	for _, clusterDir := range clustersDir {
+		lvl1Dirs, err := os.ReadDir(filepath.Join(archive, clusterDir.Name()))
+		if err != nil {
+			return err
+		}
+
+		for _, lvl1Dir := range lvl1Dirs {
+			if !lvl1Dir.IsDir() {
+				// Could be the cluster.json file
+				continue
+			}
+
+			lvl2Dirs, err := os.ReadDir(filepath.Join(archive, clusterDir.Name(), lvl1Dir.Name()))
+			if err != nil {
+				return err
+			}
+
+			for _, lvl2Dir := range lvl2Dirs {
+				dirpath := filepath.Join(archive, clusterDir.Name(), lvl1Dir.Name(), lvl2Dir.Name())
+				startTimeDirs, err := os.ReadDir(dirpath)
+				if err != nil {
+					return err
+				}
+
+				// For compability with the old job-archive directory structure where
+				// there was no start time directory.
+				for _, startTimeDir := range startTimeDirs {
+					if startTimeDir.Type().IsRegular() && startTimeDir.Name() == "meta.json" {
+						if err := handleDirectory(dirpath); err != nil {
+							log.Errorf("in %s: %s", dirpath, err.Error())
+						}
+					} else if startTimeDir.IsDir() {
+						if err := handleDirectory(filepath.Join(dirpath, startTimeDir.Name())); err != nil {
+							log.Errorf("in %s: %s", filepath.Join(dirpath, startTimeDir.Name()), err.Error())
+						}
+					}
+				}
+			}
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return err
+	}
+
+	// Create indexes after inserts so that they do not
+	// need to be continually updated.
+	if _, err := db.Exec(JobsDbIndexes); err != nil {
+		return err
+	}
+
+	log.Printf("A total of %d jobs have been registered in %.3f seconds.\n", i, time.Since(starttime).Seconds())
+	return nil
+}
+
+// TODO: Remove double logic, use repository/import.go!
+// Read the `meta.json` file at `path` and insert it to the database using the prepared
+// insert statement `stmt`. `tags` maps all existing tags to their database ID.
+func loadJob(tx *sqlx.Tx, stmt *sqlx.NamedStmt, tags map[string]int64, path string) error {
+	f, err := os.Open(filepath.Join(path, "meta.json"))
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	jobMeta := schema.JobMeta{BaseJob: schema.JobDefaults}
+	if err := json.NewDecoder(bufio.NewReader(f)).Decode(&jobMeta); err != nil {
+		return err
+	}
+
+	jobMeta.MonitoringStatus = schema.MonitoringStatusArchivingSuccessful
+	job := schema.Job{
+		BaseJob:       jobMeta.BaseJob,
+		StartTime:     time.Unix(jobMeta.StartTime, 0),
+		StartTimeUnix: jobMeta.StartTime,
+	}
+
+	// TODO: Other metrics...
+	job.FlopsAnyAvg = loadJobStat(&jobMeta, "flops_any")
+	job.MemBwAvg = loadJobStat(&jobMeta, "mem_bw")
+	job.NetBwAvg = loadJobStat(&jobMeta, "net_bw")
+	job.FileBwAvg = loadJobStat(&jobMeta, "file_bw")
+
+	job.RawResources, err = json.Marshal(job.Resources)
+	if err != nil {
+		return err
+	}
+
+	job.RawMetaData, err = json.Marshal(job.MetaData)
+	if err != nil {
+		return err
+	}
+
+	if err := SanityChecks(&job.BaseJob); err != nil {
+		return err
+	}
+
+	res, err := stmt.Exec(job)
+	if err != nil {
+		return err
+	}
+
+	id, err := res.LastInsertId()
+	if err != nil {
+		return err
+	}
+
+	for _, tag := range job.Tags {
+		tagstr := tag.Name + ":" + tag.Type
+		tagId, ok := tags[tagstr]
+		if !ok {
+			res, err := tx.Exec(`INSERT INTO tag (tag_name, tag_type) VALUES (?, ?)`, tag.Name, tag.Type)
+			if err != nil {
+				return err
+			}
+			tagId, err = res.LastInsertId()
+			if err != nil {
+				return err
+			}
+			tags[tagstr] = tagId
+		}
+
+		if _, err := tx.Exec(`INSERT INTO jobtag (job_id, tag_id) VALUES (?, ?)`, id, tagId); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

internal/repository/job.go

@@ -0,0 +1,411 @@
+package repository
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/ClusterCockpit/cc-backend/internal/auth"
+	"github.com/ClusterCockpit/cc-backend/internal/graph/model"
+	"github.com/ClusterCockpit/cc-backend/pkg/log"
+	"github.com/ClusterCockpit/cc-backend/pkg/schema"
+	sq "github.com/Masterminds/squirrel"
+	"github.com/iamlouk/lrucache"
+	"github.com/jmoiron/sqlx"
+)
+
+var (
+	jobRepoOnce     sync.Once
+	jobRepoInstance *JobRepository
+)
+
+type JobRepository struct {
+	DB *sqlx.DB
+
+	stmtCache *sq.StmtCache
+	cache     *lrucache.Cache
+}
+
+func GetRepository() *JobRepository {
+	jobRepoOnce.Do(func() {
+		db := GetConnection()
+
+		jobRepoInstance = &JobRepository{
+			DB:        db.DB,
+			stmtCache: sq.NewStmtCache(db.DB),
+			cache:     lrucache.New(1024 * 1024),
+		}
+	})
+
+	return jobRepoInstance
+}
+
+var jobColumns []string = []string{
+	"job.id", "job.job_id", "job.user", "job.project", "job.cluster", "job.subcluster", "job.start_time", "job.partition", "job.array_job_id",
+	"job.num_nodes", "job.num_hwthreads", "job.num_acc", "job.exclusive", "job.monitoring_status", "job.smt", "job.job_state",
+	"job.duration", "job.walltime", "job.resources", // "job.meta_data",
+}
+
+func scanJob(row interface{ Scan(...interface{}) error }) (*schema.Job, error) {
+	job := &schema.Job{}
+	if err := row.Scan(
+		&job.ID, &job.JobID, &job.User, &job.Project, &job.Cluster, &job.SubCluster, &job.StartTimeUnix, &job.Partition, &job.ArrayJobId,
+		&job.NumNodes, &job.NumHWThreads, &job.NumAcc, &job.Exclusive, &job.MonitoringStatus, &job.SMT, &job.State,
+		&job.Duration, &job.Walltime, &job.RawResources /*&job.MetaData*/); err != nil {
+		return nil, err
+	}
+
+	if err := json.Unmarshal(job.RawResources, &job.Resources); err != nil {
+		return nil, err
+	}
+
+	job.StartTime = time.Unix(job.StartTimeUnix, 0)
+	if job.Duration == 0 && job.State == schema.JobStateRunning {
+		job.Duration = int32(time.Since(job.StartTime).Seconds())
+	}
+
+	job.RawResources = nil
+	return job, nil
+}
+
+func (r *JobRepository) FetchMetadata(job *schema.Job) (map[string]string, error) {
+	cachekey := fmt.Sprintf("metadata:%d", job.ID)
+	if cached := r.cache.Get(cachekey, nil); cached != nil {
+		job.MetaData = cached.(map[string]string)
+		return job.MetaData, nil
+	}
+
+	if err := sq.Select("job.meta_data").From("job").Where("job.id = ?", job.ID).
+		RunWith(r.stmtCache).QueryRow().Scan(&job.RawMetaData); err != nil {
+		return nil, err
+	}
+
+	if len(job.RawMetaData) == 0 {
+		return nil, nil
+	}
+
+	if err := json.Unmarshal(job.RawMetaData, &job.MetaData); err != nil {
+		return nil, err
+	}
+
+	r.cache.Put(cachekey, job.MetaData, len(job.RawMetaData), 24*time.Hour)
+	return job.MetaData, nil
+}
+
+func (r *JobRepository) UpdateMetadata(job *schema.Job, key, val string) (err error) {
+	cachekey := fmt.Sprintf("metadata:%d", job.ID)
+	r.cache.Del(cachekey)
+	if job.MetaData == nil {
+		if _, err = r.FetchMetadata(job); err != nil {
+			return err
+		}
+	}
+
+	if job.MetaData != nil {
+		cpy := make(map[string]string, len(job.MetaData)+1)
+		for k, v := range job.MetaData {
+			cpy[k] = v
+		}
+		cpy[key] = val
+		job.MetaData = cpy
+	} else {
+		job.MetaData = map[string]string{key: val}
+	}
+
+	if job.RawMetaData, err = json.Marshal(job.MetaData); err != nil {
+		return err
+	}
+
+	if _, err = sq.Update("job").Set("meta_data", job.RawMetaData).Where("job.id = ?", job.ID).RunWith(r.stmtCache).Exec(); err != nil {
+		return err
+	}
+
+	r.cache.Put(cachekey, job.MetaData, len(job.RawMetaData), 24*time.Hour)
+	return nil
+}
+
+// Find executes a SQL query to find a specific batch job.
+// The job is queried using the batch job id, the cluster name,
+// and the start time of the job in UNIX epoch time seconds.
+// It returns a pointer to a schema.Job data structure and an error variable.
+// To check if no job was found test err == sql.ErrNoRows
+func (r *JobRepository) Find(
+	jobId *int64,
+	cluster *string,
+	startTime *int64) (*schema.Job, error) {
+
+	q := sq.Select(jobColumns...).From("job").
+		Where("job.job_id = ?", *jobId)
+
+	if cluster != nil {
+		q = q.Where("job.cluster = ?", *cluster)
+	}
+	if startTime != nil {
+		q = q.Where("job.start_time = ?", *startTime)
+	}
+
+	return scanJob(q.RunWith(r.stmtCache).QueryRow())
+}
+
+// FindById executes a SQL query to find a specific batch job.
+// The job is queried using the database id.
+// It returns a pointer to a schema.Job data structure and an error variable.
+// To check if no job was found test err == sql.ErrNoRows
+func (r *JobRepository) FindById(
+	jobId int64) (*schema.Job, error) {
+	q := sq.Select(jobColumns...).
+		From("job").Where("job.id = ?", jobId)
+	return scanJob(q.RunWith(r.stmtCache).QueryRow())
+}
+
+// Start inserts a new job in the table, returning the unique job ID.
+// Statistics are not transfered!
+func (r *JobRepository) Start(job *schema.JobMeta) (id int64, err error) {
+	job.RawResources, err = json.Marshal(job.Resources)
+	if err != nil {
+		return -1, fmt.Errorf("encoding resources field failed: %w", err)
+	}
+
+	job.RawMetaData, err = json.Marshal(job.MetaData)
+	if err != nil {
+		return -1, fmt.Errorf("encoding metaData field failed: %w", err)
+	}
+
+	res, err := r.DB.NamedExec(`INSERT INTO job (
+		job_id, user, project, cluster, subcluster, `+"`partition`"+`, array_job_id, num_nodes, num_hwthreads, num_acc,
+		exclusive, monitoring_status, smt, job_state, start_time, duration, walltime, resources, meta_data
+	) VALUES (
+		:job_id, :user, :project, :cluster, :subcluster, :partition, :array_job_id, :num_nodes, :num_hwthreads, :num_acc,
+		:exclusive, :monitoring_status, :smt, :job_state, :start_time, :duration, :walltime, :resources, :meta_data
+	);`, job)
+	if err != nil {
+		return -1, err
+	}
+
+	return res.LastInsertId()
+}
+
+// Stop updates the job with the database id jobId using the provided arguments.
+func (r *JobRepository) Stop(
+	jobId int64,
+	duration int32,
+	state schema.JobState,
+	monitoringStatus int32) (err error) {
+
+	stmt := sq.Update("job").
+		Set("job_state", state).
+		Set("duration", duration).
+		Set("monitoring_status", monitoringStatus).
+		Where("job.id = ?", jobId)
+
+	_, err = stmt.RunWith(r.stmtCache).Exec()
+	return
+}
+
+// TODO: Use node hours instead: SELECT job.user, sum(job.num_nodes * (CASE WHEN job.job_state = "running" THEN CAST(strftime('%s', 'now') AS INTEGER) - job.start_time ELSE job.duration END)) as x FROM job GROUP BY user ORDER BY x DESC;
+func (r *JobRepository) CountGroupedJobs(ctx context.Context, aggreg model.Aggregate, filters []*model.JobFilter, weight *model.Weights, limit *int) (map[string]int, error) {
+	if !aggreg.IsValid() {
+		return nil, errors.New("invalid aggregate")
+	}
+
+	runner := (sq.BaseRunner)(r.stmtCache)
+	count := "count(*) as count"
+	if weight != nil {
+		switch *weight {
+		case model.WeightsNodeCount:
+			count = "sum(job.num_nodes) as count"
+		case model.WeightsNodeHours:
+			now := time.Now().Unix()
+			count = fmt.Sprintf(`sum(job.num_nodes * (CASE WHEN job.job_state = "running" THEN %d - job.start_time ELSE job.duration END)) as count`, now)
+			runner = r.DB
+		}
+	}
+
+	q := sq.Select("job."+string(aggreg), count).From("job").GroupBy("job." + string(aggreg)).OrderBy("count DESC")
+	q = SecurityCheck(ctx, q)
+	for _, f := range filters {
+		q = BuildWhereClause(f, q)
+	}
+	if limit != nil {
+		q = q.Limit(uint64(*limit))
+	}
+
+	counts := map[string]int{}
+	rows, err := q.RunWith(runner).Query()
+	if err != nil {
+		return nil, err
+	}
+
+	for rows.Next() {
+		var group string
+		var count int
+		if err := rows.Scan(&group, &count); err != nil {
+			return nil, err
+		}
+
+		counts[group] = count
+	}
+
+	return counts, nil
+}
+
+func (r *JobRepository) UpdateMonitoringStatus(job int64, monitoringStatus int32) (err error) {
+	stmt := sq.Update("job").
+		Set("monitoring_status", monitoringStatus).
+		Where("job.id = ?", job)
+
+	_, err = stmt.RunWith(r.stmtCache).Exec()
+	return
+}
+
+// Stop updates the job with the database id jobId using the provided arguments.
+func (r *JobRepository) Archive(
+	jobId int64,
+	monitoringStatus int32,
+	metricStats map[string]schema.JobStatistics) error {
+
+	stmt := sq.Update("job").
+		Set("monitoring_status", monitoringStatus).
+		Where("job.id = ?", jobId)
+
+	for metric, stats := range metricStats {
+		switch metric {
+		case "flops_any":
+			stmt = stmt.Set("flops_any_avg", stats.Avg)
+		case "mem_used":
+			stmt = stmt.Set("mem_used_max", stats.Max)
+		case "mem_bw":
+			stmt = stmt.Set("mem_bw_avg", stats.Avg)
+		case "load":
+			stmt = stmt.Set("load_avg", stats.Avg)
+		case "net_bw":
+			stmt = stmt.Set("net_bw_avg", stats.Avg)
+		case "file_bw":
+			stmt = stmt.Set("file_bw_avg", stats.Avg)
+		}
+	}
+
+	if _, err := stmt.RunWith(r.stmtCache).Exec(); err != nil {
+		return err
+	}
+	return nil
+}
+
+var ErrNotFound = errors.New("no such job or user")
+
+// FindJobOrUser returns a job database ID or a username if a job or user machtes the search term.
+// As 0 is a valid job id, check if username is "" instead in order to check what machted.
+// If nothing matches the search, `ErrNotFound` is returned.
+func (r *JobRepository) FindJobOrUser(ctx context.Context, searchterm string) (job int64, username string, err error) {
+	user := auth.GetUser(ctx)
+	if id, err := strconv.Atoi(searchterm); err == nil {
+		qb := sq.Select("job.id").From("job").Where("job.job_id = ?", id)
+		if user != nil && !user.HasRole(auth.RoleAdmin) {
+			qb = qb.Where("job.user = ?", user.Username)
+		}
+
+		err := qb.RunWith(r.stmtCache).QueryRow().Scan(&job)
+		if err != nil && err != sql.ErrNoRows {
+			return 0, "", err
+		} else if err == nil {
+			return job, "", nil
+		}
+	}
+
+	if user == nil || user.HasRole(auth.RoleAdmin) {
+		err := sq.Select("job.user").Distinct().From("job").
+			Where("job.user = ?", searchterm).
+			RunWith(r.stmtCache).QueryRow().Scan(&username)
+		if err != nil && err != sql.ErrNoRows {
+			return 0, "", err
+		} else if err == nil {
+			return 0, username, nil
+		}
+	}
+
+	return 0, "", ErrNotFound
+}
+
+func (r *JobRepository) Partitions(cluster string) ([]string, error) {
+	var err error
+	partitions := r.cache.Get("partitions:"+cluster, func() (interface{}, time.Duration, int) {
+		parts := []string{}
+		if err = r.DB.Select(&parts, `SELECT DISTINCT job.partition FROM job WHERE job.cluster = ?;`, cluster); err != nil {
+			return nil, 0, 1000
+		}
+
+		return parts, 1 * time.Hour, 1
+	})
+	if err != nil {
+		return nil, err
+	}
+	return partitions.([]string), nil
+}
+
+// AllocatedNodes returns a map of all subclusters to a map of hostnames to the amount of jobs running on that host.
+// Hosts with zero jobs running on them will not show up!
+func (r *JobRepository) AllocatedNodes(cluster string) (map[string]map[string]int, error) {
+	subclusters := make(map[string]map[string]int)
+	rows, err := sq.Select("resources", "subcluster").From("job").
+		Where("job.job_state = 'running'").
+		Where("job.cluster = ?", cluster).
+		RunWith(r.stmtCache).Query()
+	if err != nil {
+		return nil, err
+	}
+
+	var raw []byte
+	defer rows.Close()
+	for rows.Next() {
+		raw = raw[0:0]
+		var resources []*schema.Resource
+		var subcluster string
+		if err := rows.Scan(&raw, &subcluster); err != nil {
+			return nil, err
+		}
+		if err := json.Unmarshal(raw, &resources); err != nil {
+			return nil, err
+		}
+
+		hosts, ok := subclusters[subcluster]
+		if !ok {
+			hosts = make(map[string]int)
+			subclusters[subcluster] = hosts
+		}
+
+		for _, resource := range resources {
+			hosts[resource.Hostname] += 1
+		}
+	}
+
+	return subclusters, nil
+}
+
+func (r *JobRepository) StopJobsExceedingWalltimeBy(seconds int) error {
+	res, err := sq.Update("job").
+		Set("monitoring_status", schema.MonitoringStatusArchivingFailed).
+		Set("duration", 0).
+		Set("job_state", schema.JobStateFailed).
+		Where("job.job_state = 'running'").
+		Where("job.walltime > 0").
+		Where(fmt.Sprintf("(%d - job.start_time) > (job.walltime + %d)", time.Now().Unix(), seconds)).
+		RunWith(r.DB).Exec()
+	if err != nil {
+		return err
+	}
+
+	rowsAffected, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+
+	if rowsAffected > 0 {
+		log.Warnf("%d jobs have been marked as failed due to running too long", rowsAffected)
+	}
+	return nil
+}

internal/repository/job_test.go

@@ -0,0 +1,66 @@
+package repository
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/jmoiron/sqlx"
+	_ "github.com/mattn/go-sqlite3"
+)
+
+var db *sqlx.DB
+
+func init() {
+	Connect("sqlite3", "../../test/test.db")
+}
+
+func setup(t *testing.T) *JobRepository {
+	return GetRepository()
+}
+
+func TestFind(t *testing.T) {
+	r := setup(t)
+
+	jobId, cluster, startTime := int64(1404396), "emmy", int64(1609299584)
+	job, err := r.Find(&jobId, &cluster, &startTime)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// fmt.Printf("%+v", job)
+
+	if job.ID != 1366 {
+		t.Errorf("wrong summary for diagnostic 3\ngot: %d \nwant: 1366", job.JobID)
+	}
+}
+
+func TestFindById(t *testing.T) {
+	r := setup(t)
+
+	job, err := r.FindById(1366)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// fmt.Printf("%+v", job)
+
+	if job.JobID != 1404396 {
+		t.Errorf("wrong summary for diagnostic 3\ngot: %d \nwant: 1404396", job.JobID)
+	}
+}
+
+func TestGetTags(t *testing.T) {
+	r := setup(t)
+
+	tags, counts, err := r.CountTags(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	fmt.Printf("TAGS %+v \n", tags)
+	// fmt.Printf("COUNTS %+v \n", counts)
+
+	if counts["bandwidth"] != 6 {
+		t.Errorf("wrong summary for diagnostic 3\ngot: %d \nwant: 6", counts["load-imbalance"])
+	}
+}

internal/repository/query.go

@@ -0,0 +1,217 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/ClusterCockpit/cc-backend/internal/auth"
+	"github.com/ClusterCockpit/cc-backend/internal/graph/model"
+	"github.com/ClusterCockpit/cc-backend/pkg/log"
+	"github.com/ClusterCockpit/cc-backend/pkg/schema"
+	sq "github.com/Masterminds/squirrel"
+)
+
+// QueryJobs returns a list of jobs matching the provided filters. page and order are optional-
+func (r *JobRepository) QueryJobs(
+	ctx context.Context,
+	filters []*model.JobFilter,
+	page *model.PageRequest,
+	order *model.OrderByInput) ([]*schema.Job, error) {
+
+	query := sq.Select(jobColumns...).From("job")
+	query = SecurityCheck(ctx, query)
+
+	if order != nil {
+		field := toSnakeCase(order.Field)
+		if order.Order == model.SortDirectionEnumAsc {
+			query = query.OrderBy(fmt.Sprintf("job.%s ASC", field))
+		} else if order.Order == model.SortDirectionEnumDesc {
+			query = query.OrderBy(fmt.Sprintf("job.%s DESC", field))
+		} else {
+			return nil, errors.New("invalid sorting order")
+		}
+	}
+
+	if page != nil && page.ItemsPerPage != -1 {
+		limit := uint64(page.ItemsPerPage)
+		query = query.Offset((uint64(page.Page) - 1) * limit).Limit(limit)
+	}
+
+	for _, f := range filters {
+		query = BuildWhereClause(f, query)
+	}
+
+	sql, args, err := query.ToSql()
+	if err != nil {
+		return nil, err
+	}
+
+	log.Debugf("SQL query: `%s`, args: %#v", sql, args)
+	rows, err := query.RunWith(r.stmtCache).Query()
+	if err != nil {
+		return nil, err
+	}
+
+	jobs := make([]*schema.Job, 0, 50)
+	for rows.Next() {
+		job, err := scanJob(rows)
+		if err != nil {
+			rows.Close()
+			return nil, err
+		}
+		jobs = append(jobs, job)
+	}
+
+	return jobs, nil
+}
+
+// CountJobs counts the number of jobs matching the filters.
+func (r *JobRepository) CountJobs(
+	ctx context.Context,
+	filters []*model.JobFilter) (int, error) {
+
+	// count all jobs:
+	query := sq.Select("count(*)").From("job")
+	query = SecurityCheck(ctx, query)
+	for _, f := range filters {
+		query = BuildWhereClause(f, query)
+	}
+	var count int
+	if err := query.RunWith(r.DB).Scan(&count); err != nil {
+		return 0, err
+	}
+
+	return count, nil
+}
+
+func SecurityCheck(ctx context.Context, query sq.SelectBuilder) sq.SelectBuilder {
+	user := auth.GetUser(ctx)
+	if user == nil || user.HasRole(auth.RoleAdmin) || user.HasRole(auth.RoleApi) {
+		return query
+	}
+
+	return query.Where("job.user = ?", user.Username)
+}
+
+// Build a sq.SelectBuilder out of a schema.JobFilter.
+func BuildWhereClause(filter *model.JobFilter, query sq.SelectBuilder) sq.SelectBuilder {
+	if filter.Tags != nil {
+		query = query.Join("jobtag ON jobtag.job_id = job.id").Where(sq.Eq{"jobtag.tag_id": filter.Tags})
+	}
+	if filter.JobID != nil {
+		query = buildStringCondition("job.job_id", filter.JobID, query)
+	}
+	if filter.ArrayJobID != nil {
+		query = query.Where("job.array_job_id = ?", *filter.ArrayJobID)
+	}
+	if filter.User != nil {
+		query = buildStringCondition("job.user", filter.User, query)
+	}
+	if filter.Project != nil {
+		query = buildStringCondition("job.project", filter.Project, query)
+	}
+	if filter.Cluster != nil {
+		query = buildStringCondition("job.cluster", filter.Cluster, query)
+	}
+	if filter.Partition != nil {
+		query = buildStringCondition("job.partition", filter.Partition, query)
+	}
+	if filter.StartTime != nil {
+		query = buildTimeCondition("job.start_time", filter.StartTime, query)
+	}
+	if filter.Duration != nil {
+		now := time.Now().Unix() // There does not seam to be a portable way to get the current unix timestamp accross different DBs.
+		query = query.Where("(CASE WHEN job.job_state = 'running' THEN (? - job.start_time) ELSE job.duration END) BETWEEN ? AND ?", now, filter.Duration.From, filter.Duration.To)
+	}
+	if filter.MinRunningFor != nil {
+		now := time.Now().Unix() // There does not seam to be a portable way to get the current unix timestamp accross different DBs.
+		query = query.Where("(job.job_state != 'running' OR (? - job.start_time) > ?)", now, *filter.MinRunningFor)
+	}
+	if filter.State != nil {
+		states := make([]string, len(filter.State))
+		for i, val := range filter.State {
+			states[i] = string(val)
+		}
+
+		query = query.Where(sq.Eq{"job.job_state": states})
+	}
+	if filter.NumNodes != nil {
+		query = buildIntCondition("job.num_nodes", filter.NumNodes, query)
+	}
+	if filter.NumAccelerators != nil {
+		query = buildIntCondition("job.num_acc", filter.NumAccelerators, query)
+	}
+	if filter.NumHWThreads != nil {
+		query = buildIntCondition("job.num_hwthreads", filter.NumHWThreads, query)
+	}
+	if filter.FlopsAnyAvg != nil {
+		query = buildFloatCondition("job.flops_any_avg", filter.FlopsAnyAvg, query)
+	}
+	if filter.MemBwAvg != nil {
+		query = buildFloatCondition("job.mem_bw_avg", filter.MemBwAvg, query)
+	}
+	if filter.LoadAvg != nil {
+		query = buildFloatCondition("job.load_avg", filter.LoadAvg, query)
+	}
+	if filter.MemUsedMax != nil {
+		query = buildFloatCondition("job.mem_used_max", filter.MemUsedMax, query)
+	}
+	return query
+}
+
+func buildIntCondition(field string, cond *model.IntRange, query sq.SelectBuilder) sq.SelectBuilder {
+	return query.Where(field+" BETWEEN ? AND ?", cond.From, cond.To)
+}
+
+func buildTimeCondition(field string, cond *model.TimeRange, query sq.SelectBuilder) sq.SelectBuilder {
+	if cond.From != nil && cond.To != nil {
+		return query.Where(field+" BETWEEN ? AND ?", cond.From.Unix(), cond.To.Unix())
+	} else if cond.From != nil {
+		return query.Where("? <= "+field, cond.From.Unix())
+	} else if cond.To != nil {
+		return query.Where(field+" <= ?", cond.To.Unix())
+	} else {
+		return query
+	}
+}
+
+func buildFloatCondition(field string, cond *model.FloatRange, query sq.SelectBuilder) sq.SelectBuilder {
+	return query.Where(field+" BETWEEN ? AND ?", cond.From, cond.To)
+}
+
+func buildStringCondition(field string, cond *model.StringInput, query sq.SelectBuilder) sq.SelectBuilder {
+	if cond.Eq != nil {
+		return query.Where(field+" = ?", *cond.Eq)
+	}
+	if cond.StartsWith != nil {
+		return query.Where(field+" LIKE ?", fmt.Sprint(*cond.StartsWith, "%"))
+	}
+	if cond.EndsWith != nil {
+		return query.Where(field+" LIKE ?", fmt.Sprint("%", *cond.EndsWith))
+	}
+	if cond.Contains != nil {
+		return query.Where(field+" LIKE ?", fmt.Sprint("%", *cond.Contains, "%"))
+	}
+	return query
+}
+
+var matchFirstCap = regexp.MustCompile("(.)([A-Z][a-z]+)")
+var matchAllCap = regexp.MustCompile("([a-z0-9])([A-Z])")
+
+func toSnakeCase(str string) string {
+	for _, c := range str {
+		if c == '\'' || c == '\\' {
+			panic("A hacker (probably not)!!!")
+		}
+	}
+
+	str = strings.ReplaceAll(str, "'", "")
+	str = strings.ReplaceAll(str, "\\", "")
+	snake := matchFirstCap.ReplaceAllString(str, "${1}_${2}")
+	snake = matchAllCap.ReplaceAllString(snake, "${1}_${2}")
+	return strings.ToLower(snake)
+}

internal/repository/tags.go

@@ -0,0 +1,150 @@
+package repository
+
+import (
+	"github.com/ClusterCockpit/cc-backend/internal/metricdata"
+	"github.com/ClusterCockpit/cc-backend/pkg/schema"
+	sq "github.com/Masterminds/squirrel"
+)
+
+// Add the tag with id `tagId` to the job with the database id `jobId`.
+func (r *JobRepository) AddTag(job int64, tag int64) ([]*schema.Tag, error) {
+	if _, err := r.stmtCache.Exec(`INSERT INTO jobtag (job_id, tag_id) VALUES ($1, $2)`, job, tag); err != nil {
+		return nil, err
+	}
+
+	j, err := r.FindById(job)
+	if err != nil {
+		return nil, err
+	}
+
+	tags, err := r.GetTags(&job)
+	if err != nil {
+		return nil, err
+	}
+
+	return tags, metricdata.UpdateTags(j, tags)
+}
+
+// Removes a tag from a job
+func (r *JobRepository) RemoveTag(job, tag int64) ([]*schema.Tag, error) {
+	if _, err := r.stmtCache.Exec("DELETE FROM jobtag WHERE jobtag.job_id = $1 AND jobtag.tag_id = $2", job, tag); err != nil {
+		return nil, err
+	}
+
+	j, err := r.FindById(job)
+	if err != nil {
+		return nil, err
+	}
+
+	tags, err := r.GetTags(&job)
+	if err != nil {
+		return nil, err
+	}
+
+	return tags, metricdata.UpdateTags(j, tags)
+}
+
+// CreateTag creates a new tag with the specified type and name and returns its database id.
+func (r *JobRepository) CreateTag(tagType string, tagName string) (tagId int64, err error) {
+	res, err := r.stmtCache.Exec("INSERT INTO tag (tag_type, tag_name) VALUES ($1, $2)", tagType, tagName)
+	if err != nil {
+		return 0, err
+	}
+
+	return res.LastInsertId()
+}
+
+func (r *JobRepository) CountTags(user *string) (tags []schema.Tag, counts map[string]int, err error) {
+	tags = make([]schema.Tag, 0, 100)
+	xrows, err := r.DB.Queryx("SELECT * FROM tag")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	for xrows.Next() {
+		var t schema.Tag
+		if err := xrows.StructScan(&t); err != nil {
+			return nil, nil, err
+		}
+		tags = append(tags, t)
+	}
+
+	q := sq.Select("t.tag_name, count(jt.tag_id)").
+		From("tag t").
+		LeftJoin("jobtag jt ON t.id = jt.tag_id").
+		GroupBy("t.tag_name")
+	if user != nil {
+		q = q.Where("jt.job_id IN (SELECT id FROM job WHERE job.user = ?)", *user)
+	}
+
+	rows, err := q.RunWith(r.stmtCache).Query()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	counts = make(map[string]int)
+	for rows.Next() {
+		var tagName string
+		var count int
+		if err := rows.Scan(&tagName, &count); err != nil {
+			return nil, nil, err
+		}
+		counts[tagName] = count
+	}
+	err = rows.Err()
+
+	return
+}
+
+// AddTagOrCreate adds the tag with the specified type and name to the job with the database id `jobId`.
+// If such a tag does not yet exist, it is created.
+func (r *JobRepository) AddTagOrCreate(jobId int64, tagType string, tagName string) (tagId int64, err error) {
+	tagId, exists := r.TagId(tagType, tagName)
+	if !exists {
+		tagId, err = r.CreateTag(tagType, tagName)
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	if _, err := r.AddTag(jobId, tagId); err != nil {
+		return 0, err
+	}
+
+	return tagId, nil
+}
+
+// TagId returns the database id of the tag with the specified type and name.
+func (r *JobRepository) TagId(tagType string, tagName string) (tagId int64, exists bool) {
+	exists = true
+	if err := sq.Select("id").From("tag").
+		Where("tag.tag_type = ?", tagType).Where("tag.tag_name = ?", tagName).
+		RunWith(r.stmtCache).QueryRow().Scan(&tagId); err != nil {
+		exists = false
+	}
+	return
+}
+
+// GetTags returns a list of all tags if job is nil or of the tags that the job with that database ID has.
+func (r *JobRepository) GetTags(job *int64) ([]*schema.Tag, error) {
+	q := sq.Select("id", "tag_type", "tag_name").From("tag")
+	if job != nil {
+		q = q.Join("jobtag ON jobtag.tag_id = tag.id").Where("jobtag.job_id = ?", *job)
+	}
+
+	rows, err := q.RunWith(r.stmtCache).Query()
+	if err != nil {
+		return nil, err
+	}
+
+	tags := make([]*schema.Tag, 0)
+	for rows.Next() {
+		tag := &schema.Tag{}
+		if err := rows.Scan(&tag.ID, &tag.Type, &tag.Name); err != nil {
+			return nil, err
+		}
+		tags = append(tags, tag)
+	}
+
+	return tags, nil
+}