Create tags if needed

This commit is contained in:
Lou Knauer 2022-02-08 12:49:28 +01:00
parent 0c2904d982
commit 0ca7dbb4f1
4 changed files with 59 additions and 43 deletions

View File

@ -153,15 +153,8 @@ func (api *RestApi) tagJob(rw http.ResponseWriter, r *http.Request) {
} }
for _, tag := range req { for _, tag := range req {
var tagId int64 tagId, err := api.JobRepository.AddTagOrCreate(job.ID, tag.Type, tag.Name)
exists := false if err != nil {
if exists, tagId = api.JobRepository.TagExists(tag.Type, tag.Name); exists {
http.Error(rw, fmt.Sprintf("the tag '%s:%s' does not exist", tag.Type, tag.Name), http.StatusNotFound)
return
}
if err := api.JobRepository.AddTag(job.JobID, tagId); err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError) http.Error(rw, err.Error(), http.StatusInternalServerError)
return return
} }
@ -226,18 +219,19 @@ func (api *RestApi) startJob(rw http.ResponseWriter, r *http.Request) {
return return
} }
res, err := api.JobRepository.Start(req) id, err := api.JobRepository.Start(&req)
if err != nil { if err != nil {
log.Errorf("insert into job table failed: %s", err.Error()) log.Errorf("insert into job table failed: %s", err.Error())
http.Error(rw, err.Error(), http.StatusInternalServerError) http.Error(rw, err.Error(), http.StatusInternalServerError)
return return
} }
id, err := res.LastInsertId() for _, tag := range req.Tags {
if err != nil { if _, err := api.JobRepository.AddTagOrCreate(id, tag.Type, tag.Name); err != nil {
http.Error(rw, err.Error(), http.StatusInternalServerError) http.Error(rw, err.Error(), http.StatusInternalServerError)
return return
} }
}
log.Printf("new job (id: %d): cluster=%s, jobId=%d, user=%s, startTime=%d", id, req.Cluster, req.JobID, req.User, req.StartTime) log.Printf("new job (id: %d): cluster=%s, jobId=%d, user=%s, startTime=%d", id, req.Cluster, req.JobID, req.User, req.StartTime)
rw.Header().Add("Content-Type", "application/json") rw.Header().Add("Content-Type", "application/json")

View File

@ -5,7 +5,6 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"os" "os"
@ -29,14 +28,6 @@ func setup(t *testing.T) *api.RestApi {
panic("prefer using sub-tests (`t.Run`) or implement `cleanup` before calling setup twice.") panic("prefer using sub-tests (`t.Run`) or implement `cleanup` before calling setup twice.")
} }
devNull, err := os.Open(os.DevNull)
if err != nil {
t.Fatal(err)
}
// Makes output cleaner
log.SetOutput(devNull)
const testclusterJson = `{ const testclusterJson = `{
"name": "testcluster", "name": "testcluster",
"partitions": [ "partitions": [
@ -129,7 +120,6 @@ func setup(t *testing.T) *api.RestApi {
} }
func cleanup() { func cleanup() {
log.SetOutput(os.Stderr)
// TODO: Clear all caches, reset all modules, etc... // TODO: Clear all caches, reset all modules, etc...
} }
@ -179,7 +169,7 @@ func TestRestApi(t *testing.T) {
"exclusive": 1, "exclusive": 1,
"monitoringStatus": 1, "monitoringStatus": 1,
"smt": 1, "smt": 1,
"tags": [], "tags": [{ "type": "testTagType", "name": "testTagName" }],
"resources": [ "resources": [
{ {
"hostname": "testhost", "hostname": "testhost",
@ -211,6 +201,11 @@ func TestRestApi(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
job.Tags, err = restapi.Resolver.Job().Tags(context.Background(), job)
if err != nil {
t.Fatal(err)
}
if job.JobID != 123 || if job.JobID != 123 ||
job.User != "testuser" || job.User != "testuser" ||
job.Project != "testproj" || job.Project != "testproj" ||
@ -223,12 +218,15 @@ func TestRestApi(t *testing.T) {
job.Exclusive != 1 || job.Exclusive != 1 ||
job.MonitoringStatus != 1 || job.MonitoringStatus != 1 ||
job.SMT != 1 || job.SMT != 1 ||
len(job.Tags) != 0 ||
!reflect.DeepEqual(job.Resources, []*schema.Resource{{Hostname: "testhost", HWThreads: []int{0, 1, 2, 3, 4, 5, 6, 7}}}) || !reflect.DeepEqual(job.Resources, []*schema.Resource{{Hostname: "testhost", HWThreads: []int{0, 1, 2, 3, 4, 5, 6, 7}}}) ||
job.StartTime.Unix() != 123456789 { job.StartTime.Unix() != 123456789 {
t.Fatalf("unexpected job properties: %#v", job) t.Fatalf("unexpected job properties: %#v", job)
} }
if len(job.Tags) != 1 || job.Tags[0].Type != "testTagType" || job.Tags[0].Name != "testTagName" {
t.Fatalf("unexpected tags: %#v", job.Tags)
}
dbid = res.DBID dbid = res.DBID
}); !ok { }); !ok {
return return

View File

@ -54,7 +54,8 @@ const JOBS_DB_SCHEMA string = `
CREATE TABLE tag ( CREATE TABLE tag (
id INTEGER PRIMARY KEY, id INTEGER PRIMARY KEY,
tag_type VARCHAR(255) NOT NULL, tag_type VARCHAR(255) NOT NULL,
tag_name VARCHAR(255) NOT NULL); tag_name VARCHAR(255) NOT NULL,
CONSTRAINT be_unique UNIQUE (tag_type, tag_name));
CREATE TABLE jobtag ( CREATE TABLE jobtag (
job_id INTEGER, job_id INTEGER,

View File

@ -1,8 +1,6 @@
package repository package repository
import ( import (
"database/sql"
"github.com/ClusterCockpit/cc-backend/log" "github.com/ClusterCockpit/cc-backend/log"
"github.com/ClusterCockpit/cc-backend/schema" "github.com/ClusterCockpit/cc-backend/schema"
sq "github.com/Masterminds/squirrel" sq "github.com/Masterminds/squirrel"
@ -52,17 +50,24 @@ func (r *JobRepository) FindById(
return job, err return job, err
} }
func (r *JobRepository) Start(job schema.JobMeta) (res sql.Result, err error) { // Start inserts a new job in the table, returning the unique job ID.
res, err = r.DB.NamedExec(`INSERT INTO job ( // Statistics are not transfered!
func (r *JobRepository) Start(job *schema.JobMeta) (id int64, err error) {
res, err := r.DB.NamedExec(`INSERT INTO job (
job_id, user, project, cluster, `+"`partition`"+`, array_job_id, num_nodes, num_hwthreads, num_acc, job_id, user, project, cluster, `+"`partition`"+`, array_job_id, num_nodes, num_hwthreads, num_acc,
exclusive, monitoring_status, smt, job_state, start_time, duration, resources, meta_data exclusive, monitoring_status, smt, job_state, start_time, duration, resources, meta_data
) VALUES ( ) VALUES (
:job_id, :user, :project, :cluster, :partition, :array_job_id, :num_nodes, :num_hwthreads, :num_acc, :job_id, :user, :project, :cluster, :partition, :array_job_id, :num_nodes, :num_hwthreads, :num_acc,
:exclusive, :monitoring_status, :smt, :job_state, :start_time, :duration, :resources, :meta_data :exclusive, :monitoring_status, :smt, :job_state, :start_time, :duration, :resources, :meta_data
);`, job) );`, job)
return 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( func (r *JobRepository) Stop(
jobId int64, jobId int64,
duration int32, duration int32,
@ -91,30 +96,48 @@ func (r *JobRepository) Stop(
} }
} }
sql, args, err := stmt.ToSql() if _, err := stmt.RunWith(r.DB).Exec(); err != nil {
if err != nil {
log.Errorf("archiving job (dbid: %d) failed: %s", jobId, err.Error())
}
if _, err := r.DB.Exec(sql, args...); err != nil {
log.Errorf("archiving job (dbid: %d) failed: %s", jobId, err.Error()) log.Errorf("archiving job (dbid: %d) failed: %s", jobId, err.Error())
} }
} }
// Add the tag with id `tagId` to the job with the database id `jobId`.
func (r *JobRepository) AddTag(jobId int64, tagId int64) error { func (r *JobRepository) AddTag(jobId int64, tagId int64) error {
_, err := r.DB.Exec(`INSERT INTO jobtag (job_id, tag_id) VALUES (?, ?)`, jobId, tagId) _, err := r.DB.Exec(`INSERT INTO jobtag (job_id, tag_id) VALUES (?, ?)`, jobId, tagId)
return err return err
} }
func (r *JobRepository) TagExists(tagType string, tagName string) (exists bool, tagId int64) { // 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.DB.Exec("INSERT INTO tag (tag_type, tag_name) VALUES ($1, $2)", tagType, tagName)
if err != nil {
return 0, err
}
return res.LastInsertId()
}
// 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
}
}
return tagId, r.AddTag(jobId, tagId)
}
// 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 exists = true
if err := sq.Select("id").From("tag"). if err := sq.Select("id").From("tag").
Where("tag.tag_type = ?", tagType).Where("tag.tag_name = ?", tagName). Where("tag.tag_type = ?", tagType).Where("tag.tag_name = ?", tagName).
RunWith(r.DB).QueryRow().Scan(&tagId); err != nil { RunWith(r.DB).QueryRow().Scan(&tagId); err != nil {
exists = false exists = false
return exists, tagId
} else {
return exists, tagId
} }
return
} }