Merge branch 'master' into 134-job-tagging

internal/repository/dbConnection.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/hooks.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/job.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
@@ -16,6 +16,7 @@ import (
 
 	"github.com/ClusterCockpit/cc-backend/internal/graph/model"
 	"github.com/ClusterCockpit/cc-backend/internal/metricdata"
+	"github.com/ClusterCockpit/cc-backend/pkg/archive"
 	"github.com/ClusterCockpit/cc-backend/pkg/log"
 	"github.com/ClusterCockpit/cc-backend/pkg/lrucache"
 	"github.com/ClusterCockpit/cc-backend/pkg/schema"
@@ -29,13 +30,11 @@ var (
 )
 
 type JobRepository struct {
-	DB     *sqlx.DB
-	driver string
-
-	stmtCache *sq.StmtCache
-	cache     *lrucache.Cache
-
+	DB             *sqlx.DB
+	stmtCache      *sq.StmtCache
+	cache          *lrucache.Cache
 	archiveChannel chan *schema.Job
+	driver         string
 	archivePending sync.WaitGroup
 }
 
@@ -60,7 +59,7 @@ func GetJobRepository() *JobRepository {
 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",
+	"job.duration", "job.walltime", "job.resources", "job.mem_used_max", "job.flops_any_avg", "job.mem_bw_avg", "job.load_avg", // "job.meta_data",
 }
 
 func scanJob(row interface{ Scan(...interface{}) error }) (*schema.Job, error) {
@@ -68,7 +67,7 @@ func scanJob(row interface{ Scan(...interface{}) error }) (*schema.Job, error) {
 	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.RawMetaData*/); err != nil {
+		&job.Duration, &job.Walltime, &job.RawResources, &job.MemUsedMax, &job.FlopsAnyAvg, &job.MemBwAvg, &job.LoadAvg /*&job.RawMetaData*/); err != nil {
 		log.Warnf("Error while scanning rows (Job): %v", err)
 		return nil, err
 	}
@@ -212,7 +211,7 @@ func (r *JobRepository) UpdateMetadata(job *schema.Job, key, val string) (err er
 	}
 
 	r.cache.Put(cachekey, job.MetaData, len(job.RawMetaData), 24*time.Hour)
-	return nil
+	return archive.UpdateMetadata(job, job.MetaData)
 }
 
 // Find executes a SQL query to find a specific batch job.
@@ -223,8 +222,8 @@ func (r *JobRepository) UpdateMetadata(job *schema.Job, key, val string) (err er
 func (r *JobRepository) Find(
 	jobId *int64,
 	cluster *string,
-	startTime *int64) (*schema.Job, error) {
-
+	startTime *int64,
+) (*schema.Job, error) {
 	start := time.Now()
 	q := sq.Select(jobColumns...).From("job").
 		Where("job.job_id = ?", *jobId)
@@ -248,8 +247,8 @@ func (r *JobRepository) Find(
 func (r *JobRepository) FindAll(
 	jobId *int64,
 	cluster *string,
-	startTime *int64) ([]*schema.Job, error) {
-
+	startTime *int64,
+) ([]*schema.Job, error) {
 	start := time.Now()
 	q := sq.Select(jobColumns...).From("job").
 		Where("job.job_id = ?", *jobId)
@@ -292,7 +291,8 @@ func (r *JobRepository) FindById(jobId int64) (*schema.Job, error) {
 
 func (r *JobRepository) FindConcurrentJobs(
 	ctx context.Context,
-	job *schema.Job) (*model.JobLinkResultList, error) {
+	job *schema.Job,
+) (*model.JobLinkResultList, error) {
 	if job == nil {
 		return nil, nil
 	}
@@ -420,8 +420,8 @@ func (r *JobRepository) Stop(
 	jobId int64,
 	duration int32,
 	state schema.JobState,
-	monitoringStatus int32) (err error) {
-
+	monitoringStatus int32,
+) (err error) {
 	stmt := sq.Update("job").
 		Set("job_state", state).
 		Set("duration", duration).
@@ -434,11 +434,14 @@ func (r *JobRepository) Stop(
 
 func (r *JobRepository) DeleteJobsBefore(startTime int64) (int, error) {
 	var cnt int
-	qs := fmt.Sprintf("SELECT count(*) FROM job WHERE job.start_time < %d", startTime)
-	err := r.DB.Get(&cnt, qs) //ignore error as it will also occur in delete statement
-	_, err = r.DB.Exec(`DELETE FROM job WHERE job.start_time < ?`, startTime)
+	q := sq.Select("count(*)").From("job").Where("job.start_time < ?", startTime)
+	q.RunWith(r.DB).QueryRow().Scan(cnt)
+	qd := sq.Delete("job").Where("job.start_time < ?", startTime)
+	_, err := qd.RunWith(r.DB).Exec()
+
 	if err != nil {
-		log.Errorf(" DeleteJobsBefore(%d): error %#v", startTime, err)
+		s, _, _ := qd.ToSql()
+		log.Errorf(" DeleteJobsBefore(%d) with %s: error %#v", startTime, s, err)
 	} else {
 		log.Debugf("DeleteJobsBefore(%d): Deleted %d jobs", startTime, cnt)
 	}
@@ -446,9 +449,12 @@ func (r *JobRepository) DeleteJobsBefore(startTime int64) (int, error) {
 }
 
 func (r *JobRepository) DeleteJobById(id int64) error {
-	_, err := r.DB.Exec(`DELETE FROM job WHERE job.id = ?`, id)
+	qd := sq.Delete("job").Where("job.id = ?", id)
+	_, err := qd.RunWith(r.DB).Exec()
+
 	if err != nil {
-		log.Errorf("DeleteJobById(%d): error %#v", id, err)
+		s, _, _ := qd.ToSql()
+		log.Errorf("DeleteJobById(%d) with %s : error %#v", id, s, err)
 	} else {
 		log.Debugf("DeleteJobById(%d): Success", id)
 	}
@@ -468,8 +474,8 @@ func (r *JobRepository) UpdateMonitoringStatus(job int64, monitoringStatus int32
 func (r *JobRepository) MarkArchived(
 	jobId int64,
 	monitoringStatus int32,
-	metricStats map[string]schema.JobStatistics) error {
-
+	metricStats map[string]schema.JobStatistics,
+) error {
 	stmt := sq.Update("job").
 		Set("monitoring_status", monitoringStatus).
 		Where("job.id = ?", jobId)
@@ -483,6 +489,7 @@ func (r *JobRepository) MarkArchived(
 		case "mem_bw":
 			stmt = stmt.Set("mem_bw_avg", stats.Avg)
 		case "load":
+			stmt = stmt.Set("load_avg", stats.Avg)
 		case "cpu_load":
 			stmt = stmt.Set("load_avg", stats.Avg)
 		case "net_bw":
@@ -577,8 +584,10 @@ func (r *JobRepository) FindUserOrProjectOrJobname(user *schema.User, searchterm
 	}
 }
 
-var ErrNotFound = errors.New("no such jobname, project or user")
-var ErrForbidden = errors.New("not authorized")
+var (
+	ErrNotFound  = errors.New("no such jobname, project or user")
+	ErrForbidden = errors.New("not authorized")
+)
 
 func (r *JobRepository) FindColumnValue(user *schema.User, searchterm string, table string, selectColumn string, whereColumn string, isLike bool) (result string, err error) {
 	compareStr := " = ?"
@@ -662,7 +671,6 @@ func (r *JobRepository) Partitions(cluster string) ([]string, error) {
 // 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) {
-
 	start := time.Now()
 	subclusters := make(map[string]map[string]int)
 	rows, err := sq.Select("resources", "subcluster").From("job").
@@ -705,7 +713,6 @@ func (r *JobRepository) AllocatedNodes(cluster string) (map[string]map[string]in
 }
 
 func (r *JobRepository) StopJobsExceedingWalltimeBy(seconds int) error {
-
 	start := time.Now()
 	res, err := sq.Update("job").
 		Set("monitoring_status", schema.MonitoringStatusArchivingFailed).
@@ -734,7 +741,6 @@ func (r *JobRepository) StopJobsExceedingWalltimeBy(seconds int) error {
 }
 
 func (r *JobRepository) FindJobsBetween(startTimeBegin int64, startTimeEnd int64) ([]*schema.Job, error) {
-
 	var query sq.SelectBuilder
 
 	if startTimeBegin == startTimeEnd || startTimeBegin > startTimeEnd {

internal/repository/job_test.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/migration.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
@@ -16,7 +16,7 @@ import (
 	"github.com/golang-migrate/migrate/v4/source/iofs"
 )
 
-const Version uint = 6
+const Version uint = 7
 
 //go:embed migrations/*
 var migrationFiles embed.FS
@@ -57,7 +57,7 @@ func checkDBVersion(backend string, db *sql.DB) error {
 		log.Fatalf("unsupported database backend: %s", backend)
 	}
 
-	v, _, err := m.Version()
+	v, dirty, err := m.Version()
 	if err != nil {
 		if err == migrate.ErrNilVersion {
 			log.Warn("Legacy database without version or missing database file!")
@@ -68,18 +68,18 @@ func checkDBVersion(backend string, db *sql.DB) error {
 
 	if v < Version {
 		return fmt.Errorf("unsupported database version %d, need %d.\nPlease backup your database file and run cc-backend -migrate-db", v, Version)
+	} else if v > Version {
+		return fmt.Errorf("unsupported database version %d, need %d.\nPlease refer to documentation how to downgrade db with external migrate tool", v, Version)
 	}
 
-	if v > Version {
-		return fmt.Errorf("unsupported database version %d, need %d.\nPlease refer to documentation how to downgrade db with external migrate tool", v, Version)
+	if dirty {
+		return fmt.Errorf("last migration to version %d has failed, please fix the db manually and force version with -force-db flag", Version)
 	}
 
 	return nil
 }
 
-func MigrateDB(backend string, db string) error {
-	var m *migrate.Migrate
-
+func getMigrateInstance(backend string, db string) (m *migrate.Migrate, err error) {
 	switch backend {
 	case "sqlite3":
 		d, err := iofs.New(migrationFiles, "migrations/sqlite3")
@@ -89,22 +89,31 @@ func MigrateDB(backend string, db string) error {
 
 		m, err = migrate.NewWithSourceInstance("iofs", d, fmt.Sprintf("sqlite3://%s?_foreign_keys=on", db))
 		if err != nil {
-			return err
+			return m, err
 		}
 	case "mysql":
 		d, err := iofs.New(migrationFiles, "migrations/mysql")
 		if err != nil {
-			return err
+			return m, err
 		}
 
 		m, err = migrate.NewWithSourceInstance("iofs", d, fmt.Sprintf("mysql://%s?multiStatements=true", db))
 		if err != nil {
-			return err
+			return m, err
 		}
 	default:
 		log.Fatalf("unsupported database backend: %s", backend)
 	}
 
+	return m, nil
+}
+
+func MigrateDB(backend string, db string) error {
+	m, err := getMigrateInstance(backend, db)
+	if err != nil {
+		return err
+	}
+
 	if err := m.Up(); err != nil {
 		if err == migrate.ErrNoChange {
 			log.Info("DB already up to date!")
@@ -116,3 +125,35 @@ func MigrateDB(backend string, db string) error {
 	m.Close()
 	return nil
 }
+
+func RevertDB(backend string, db string) error {
+	m, err := getMigrateInstance(backend, db)
+	if err != nil {
+		return err
+	}
+
+	if err := m.Migrate(Version - 1); err != nil {
+		if err == migrate.ErrNoChange {
+			log.Info("DB already up to date!")
+		} else {
+			return err
+		}
+	}
+
+	m.Close()
+	return nil
+}
+
+func ForceDB(backend string, db string) error {
+	m, err := getMigrateInstance(backend, db)
+	if err != nil {
+		return err
+	}
+
+	if err := m.Force(int(Version)); err != nil {
+		return err
+	}
+
+	m.Close()
+	return nil
+}

internal/repository/migrations/mysql/07_fix-tag-id.down.sql

@@ -0,0 +1,3 @@
+SET FOREIGN_KEY_CHECKS = 0;
+ALTER TABLE tag MODIFY id INTEGER;
+SET FOREIGN_KEY_CHECKS = 1;

internal/repository/migrations/mysql/07_fix-tag-id.up.sql

@@ -0,0 +1,3 @@
+SET FOREIGN_KEY_CHECKS = 0;
+ALTER TABLE tag MODIFY id INTEGER AUTO_INCREMENT;
+SET FOREIGN_KEY_CHECKS = 1;

internal/repository/query.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
@@ -96,7 +96,7 @@ func SecurityCheck(ctx context.Context, query sq.SelectBuilder) (sq.SelectBuilde
 	user := GetUserFromContext(ctx)
 	if user == nil {
 		var qnil sq.SelectBuilder
-		return qnil, fmt.Errorf("user context is nil!")
+		return qnil, fmt.Errorf("user context is nil")
 	} else if user.HasAnyRole([]schema.Role{schema.RoleAdmin, schema.RoleSupport, schema.RoleApi}) { // Admin & Co. : All jobs
 		return query, nil
 	} else if user.HasRole(schema.RoleManager) { // Manager : Add filter for managed projects' jobs only + personal jobs

internal/repository/repository_test.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/stats.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
@@ -8,11 +8,15 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"math"
 	"time"
 
 	"github.com/ClusterCockpit/cc-backend/internal/config"
 	"github.com/ClusterCockpit/cc-backend/internal/graph/model"
+	"github.com/ClusterCockpit/cc-backend/internal/metricdata"
+	"github.com/ClusterCockpit/cc-backend/pkg/archive"
 	"github.com/ClusterCockpit/cc-backend/pkg/log"
+	"github.com/ClusterCockpit/cc-backend/pkg/schema"
 	sq "github.com/Masterminds/squirrel"
 )
 
@@ -450,6 +454,39 @@ func (r *JobRepository) AddHistograms(
 	return stat, nil
 }
 
+// Requires thresholds for metric from config for cluster? Of all clusters and use largest? split to 10 + 1 for artifacts?
+func (r *JobRepository) AddMetricHistograms(
+	ctx context.Context,
+	filter []*model.JobFilter,
+	metrics []string,
+	stat *model.JobsStatistics) (*model.JobsStatistics, error) {
+	start := time.Now()
+
+	// Running Jobs Only: First query jobdata from sqlite, then query data and make bins
+	for _, f := range filter {
+		if f.State != nil {
+			if len(f.State) == 1 && f.State[0] == "running" {
+				stat.HistMetrics = r.runningJobsMetricStatisticsHistogram(ctx, metrics, filter)
+				log.Debugf("Timer AddMetricHistograms %s", time.Since(start))
+				return stat, nil
+			}
+		}
+	}
+
+	// All other cases: Query and make bins in sqlite directly
+	for _, m := range metrics {
+		metricHisto, err := r.jobsMetricStatisticsHistogram(ctx, m, filter)
+		if err != nil {
+			log.Warnf("Error while loading job metric statistics histogram: %s", m)
+			continue
+		}
+		stat.HistMetrics = append(stat.HistMetrics, metricHisto)
+	}
+
+	log.Debugf("Timer AddMetricHistograms %s", time.Since(start))
+	return stat, nil
+}
+
 // `value` must be the column grouped by, but renamed to "value"
 func (r *JobRepository) jobsStatisticsHistogram(
 	ctx context.Context,
@@ -487,3 +524,231 @@ func (r *JobRepository) jobsStatisticsHistogram(
 	log.Debugf("Timer jobsStatisticsHistogram %s", time.Since(start))
 	return points, nil
 }
+
+func (r *JobRepository) jobsMetricStatisticsHistogram(
+	ctx context.Context,
+	metric string,
+	filters []*model.JobFilter) (*model.MetricHistoPoints, error) {
+
+	var dbMetric string
+	switch metric {
+	case "cpu_load":
+		dbMetric = "load_avg"
+	case "flops_any":
+		dbMetric = "flops_any_avg"
+	case "mem_bw":
+		dbMetric = "mem_bw_avg"
+	case "mem_used":
+		dbMetric = "mem_used_max"
+	case "net_bw":
+		dbMetric = "net_bw_avg"
+	case "file_bw":
+		dbMetric = "file_bw_avg"
+	default:
+		return nil, fmt.Errorf("%s not implemented", metric)
+	}
+
+	// Get specific Peak or largest Peak
+	var metricConfig *schema.MetricConfig
+	var peak float64 = 0.0
+	var unit string = ""
+
+	for _, f := range filters {
+		if f.Cluster != nil {
+			metricConfig = archive.GetMetricConfig(*f.Cluster.Eq, metric)
+			peak = metricConfig.Peak
+			unit = metricConfig.Unit.Prefix + metricConfig.Unit.Base
+			log.Debugf("Cluster %s filter found with peak %f for %s", *f.Cluster.Eq, peak, metric)
+		}
+	}
+
+	if peak == 0.0 {
+		for _, c := range archive.Clusters {
+			for _, m := range c.MetricConfig {
+				if m.Name == metric {
+					if m.Peak > peak {
+						peak = m.Peak
+					}
+					if unit == "" {
+						unit = m.Unit.Prefix + m.Unit.Base
+					}
+				}
+			}
+		}
+	}
+
+	// log.Debugf("Metric %s: DB %s, Peak %f, Unit %s", metric, dbMetric, peak, unit)
+	// Make bins, see https://jereze.com/code/sql-histogram/
+
+	start := time.Now()
+
+	crossJoinQuery := sq.Select(
+		fmt.Sprintf(`max(%s) as max`, dbMetric),
+		fmt.Sprintf(`min(%s) as min`, dbMetric),
+	).From("job").Where(
+		fmt.Sprintf(`%s is not null`, dbMetric),
+	).Where(
+		fmt.Sprintf(`%s <= %f`, dbMetric, peak),
+	)
+
+	crossJoinQuery, cjqerr := SecurityCheck(ctx, crossJoinQuery)
+
+	if cjqerr != nil {
+		return nil, cjqerr
+	}
+
+	for _, f := range filters {
+		crossJoinQuery = BuildWhereClause(f, crossJoinQuery)
+	}
+
+	crossJoinQuerySql, crossJoinQueryArgs, sqlerr := crossJoinQuery.ToSql()
+	if sqlerr != nil {
+		return nil, sqlerr
+	}
+
+	bins := 10
+	binQuery := fmt.Sprintf(`CAST( (case when job.%s = value.max then value.max*0.999999999 else job.%s end - value.min) / (value.max - value.min) * %d as INTEGER )`, dbMetric, dbMetric, bins)
+
+	mainQuery := sq.Select(
+		fmt.Sprintf(`%s + 1 as bin`, binQuery),
+		fmt.Sprintf(`count(job.%s) as count`, dbMetric),
+		fmt.Sprintf(`CAST(((value.max / %d) * (%s     )) as INTEGER ) as min`, bins, binQuery),
+		fmt.Sprintf(`CAST(((value.max / %d) * (%s + 1 )) as INTEGER ) as max`, bins, binQuery),
+	).From("job").CrossJoin(
+		fmt.Sprintf(`(%s) as value`, crossJoinQuerySql), crossJoinQueryArgs...,
+	).Where(fmt.Sprintf(`job.%s is not null and job.%s <= %f`, dbMetric, dbMetric, peak))
+
+	mainQuery, qerr := SecurityCheck(ctx, mainQuery)
+
+	if qerr != nil {
+		return nil, qerr
+	}
+
+	for _, f := range filters {
+		mainQuery = BuildWhereClause(f, mainQuery)
+	}
+
+	// Finalize query with Grouping and Ordering
+	mainQuery = mainQuery.GroupBy("bin").OrderBy("bin")
+
+	rows, err := mainQuery.RunWith(r.DB).Query()
+	if err != nil {
+		log.Errorf("Error while running mainQuery: %s", err)
+		return nil, err
+	}
+
+	points := make([]*model.MetricHistoPoint, 0)
+	for rows.Next() {
+		point := model.MetricHistoPoint{}
+		if err := rows.Scan(&point.Bin, &point.Count, &point.Min, &point.Max); err != nil {
+			log.Warnf("Error while scanning rows for %s", metric)
+			return nil, err // Totally bricks cc-backend if returned and if all metrics requested?
+		}
+
+		points = append(points, &point)
+	}
+
+	result := model.MetricHistoPoints{Metric: metric, Unit: unit, Data: points}
+
+	log.Debugf("Timer jobsStatisticsHistogram %s", time.Since(start))
+	return &result, nil
+}
+
+func (r *JobRepository) runningJobsMetricStatisticsHistogram(
+	ctx context.Context,
+	metrics []string,
+	filters []*model.JobFilter) []*model.MetricHistoPoints {
+
+	// Get Jobs
+	jobs, err := r.QueryJobs(ctx, filters, &model.PageRequest{Page: 1, ItemsPerPage: 500 + 1}, nil)
+	if err != nil {
+		log.Errorf("Error while querying jobs for footprint: %s", err)
+		return nil
+	}
+	if len(jobs) > 500 {
+		log.Errorf("too many jobs matched (max: %d)", 500)
+		return nil
+	}
+
+	// Get AVGs from metric repo
+	avgs := make([][]schema.Float, len(metrics))
+	for i := range avgs {
+		avgs[i] = make([]schema.Float, 0, len(jobs))
+	}
+
+	for _, job := range jobs {
+		if job.MonitoringStatus == schema.MonitoringStatusDisabled || job.MonitoringStatus == schema.MonitoringStatusArchivingFailed {
+			continue
+		}
+
+		if err := metricdata.LoadAverages(job, metrics, avgs, ctx); err != nil {
+			log.Errorf("Error while loading averages for histogram: %s", err)
+			return nil
+		}
+	}
+
+	// Iterate metrics to fill endresult
+	data := make([]*model.MetricHistoPoints, 0)
+	for idx, metric := range metrics {
+		// Get specific Peak or largest Peak
+		var metricConfig *schema.MetricConfig
+		var peak float64 = 0.0
+		var unit string = ""
+
+		for _, f := range filters {
+			if f.Cluster != nil {
+				metricConfig = archive.GetMetricConfig(*f.Cluster.Eq, metric)
+				peak = metricConfig.Peak
+				unit = metricConfig.Unit.Prefix + metricConfig.Unit.Base
+				log.Debugf("Cluster %s filter found with peak %f for %s", *f.Cluster.Eq, peak, metric)
+			}
+		}
+
+		if peak == 0.0 {
+			for _, c := range archive.Clusters {
+				for _, m := range c.MetricConfig {
+					if m.Name == metric {
+						if m.Peak > peak {
+							peak = m.Peak
+						}
+						if unit == "" {
+							unit = m.Unit.Prefix + m.Unit.Base
+						}
+					}
+				}
+			}
+		}
+
+		// Make and fill bins
+		bins := 10.0
+		peakBin := peak / bins
+
+		points := make([]*model.MetricHistoPoint, 0)
+		for b := 0; b < 10; b++ {
+			count := 0
+			bindex := b + 1
+			bmin := math.Round(peakBin * float64(b))
+			bmax := math.Round(peakBin * (float64(b) + 1.0))
+
+			// Iterate AVG values for indexed metric and count for bins
+			for _, val := range avgs[idx] {
+				if float64(val) >= bmin && float64(val) < bmax {
+					count += 1
+				}
+			}
+
+			bminint := int(bmin)
+			bmaxint := int(bmax)
+
+			// Append Bin to Metric Result Array
+			point := model.MetricHistoPoint{Bin: &bindex, Count: count, Min: &bminint, Max: &bmaxint}
+			points = append(points, &point)
+		}
+
+		// Append Metric Result Array to final results array
+		result := model.MetricHistoPoints{Metric: metric, Unit: unit, Data: points}
+		data = append(data, &result)
+	}
+
+	return data
+}

internal/repository/stats_test.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/tags.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
@@ -15,8 +15,11 @@ import (
 
 // 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 {
-		log.Error("Error while running query")
+	q := sq.Insert("jobtag").Columns("job_id", "tag_id").Values(job, tag)
+
+	if _, err := q.RunWith(r.stmtCache).Exec(); err != nil {
+		s, _, _ := q.ToSql()
+		log.Errorf("Error adding tag with %s: %v", s, err)
 		return nil, err
 	}
 
@@ -37,8 +40,11 @@ func (r *JobRepository) AddTag(job int64, tag int64) ([]*schema.Tag, error) {
 
 // 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 {
-		log.Error("Error while running query")
+	q := sq.Delete("jobtag").Where("jobtag.job_id = ?", job).Where("jobtag.tag_id = ?", tag)
+
+	if _, err := q.RunWith(r.stmtCache).Exec(); err != nil {
+		s, _, _ := q.ToSql()
+		log.Errorf("Error adding tag with %s: %v", s, err)
 		return nil, err
 	}
 
@@ -59,8 +65,12 @@ func (r *JobRepository) RemoveTag(job, tag int64) ([]*schema.Tag, error) {
 
 // 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)
+	q := sq.Insert("tag").Columns("tag_type", "tag_name").Values(tagType, tagName)
+
+	res, err := q.RunWith(r.stmtCache).Exec()
 	if err != nil {
+		s, _, _ := q.ToSql()
+		log.Errorf("Error inserting tag with %s: %v", s, err)
 		return 0, err
 	}
 
@@ -167,7 +177,8 @@ func (r *JobRepository) GetTags(job *int64) ([]*schema.Tag, error) {
 
 	rows, err := q.RunWith(r.stmtCache).Query()
 	if err != nil {
-		log.Error("Error while running query")
+		s, _, _ := q.ToSql()
+		log.Errorf("Error get tags with %s: %v", s, err)
 		return nil, err
 	}
 

internal/repository/transaction.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/user.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2023 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/userConfig.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.

internal/repository/userConfig_test.go

@@ -1,4 +1,4 @@
-// Copyright (C) 2022 NHR@FAU, University Erlangen-Nuremberg.
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
 // All rights reserved.
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.