From 0ca7dbb4f180b3b171f188d589608986592d4b27 Mon Sep 17 00:00:00 2001 From: Lou Knauer Date: Tue, 8 Feb 2022 12:49:28 +0100 Subject: [PATCH] Create tags if needed --- api/rest.go | 22 +++++++------------ api_test.go | 22 +++++++++---------- init-db.go | 3 ++- repository/job.go | 55 +++++++++++++++++++++++++++++++++-------------- 4 files changed, 59 insertions(+), 43 deletions(-) diff --git a/api/rest.go b/api/rest.go index ad4f2ee..a871b5b 100644 --- a/api/rest.go +++ b/api/rest.go @@ -153,15 +153,8 @@ func (api *RestApi) tagJob(rw http.ResponseWriter, r *http.Request) { } for _, tag := range req { - var tagId int64 - exists := false - - 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 { + tagId, err := api.JobRepository.AddTagOrCreate(job.ID, tag.Type, tag.Name) + if err != nil { http.Error(rw, err.Error(), http.StatusInternalServerError) return } @@ -226,17 +219,18 @@ func (api *RestApi) startJob(rw http.ResponseWriter, r *http.Request) { return } - res, err := api.JobRepository.Start(req) + id, err := api.JobRepository.Start(&req) if err != nil { log.Errorf("insert into job table failed: %s", err.Error()) http.Error(rw, err.Error(), http.StatusInternalServerError) return } - id, err := res.LastInsertId() - if err != nil { - http.Error(rw, err.Error(), http.StatusInternalServerError) - return + for _, tag := range req.Tags { + if _, err := api.JobRepository.AddTagOrCreate(id, tag.Type, tag.Name); err != nil { + http.Error(rw, err.Error(), http.StatusInternalServerError) + return + } } log.Printf("new job (id: %d): cluster=%s, jobId=%d, user=%s, startTime=%d", id, req.Cluster, req.JobID, req.User, req.StartTime) diff --git a/api_test.go b/api_test.go index 537aa43..e80cd21 100644 --- a/api_test.go +++ b/api_test.go @@ -5,7 +5,6 @@ import ( "context" "encoding/json" "fmt" - "log" "net/http" "net/http/httptest" "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.") } - devNull, err := os.Open(os.DevNull) - if err != nil { - t.Fatal(err) - } - - // Makes output cleaner - log.SetOutput(devNull) - const testclusterJson = `{ "name": "testcluster", "partitions": [ @@ -129,7 +120,6 @@ func setup(t *testing.T) *api.RestApi { } func cleanup() { - log.SetOutput(os.Stderr) // TODO: Clear all caches, reset all modules, etc... } @@ -179,7 +169,7 @@ func TestRestApi(t *testing.T) { "exclusive": 1, "monitoringStatus": 1, "smt": 1, - "tags": [], + "tags": [{ "type": "testTagType", "name": "testTagName" }], "resources": [ { "hostname": "testhost", @@ -211,6 +201,11 @@ func TestRestApi(t *testing.T) { t.Fatal(err) } + job.Tags, err = restapi.Resolver.Job().Tags(context.Background(), job) + if err != nil { + t.Fatal(err) + } + if job.JobID != 123 || job.User != "testuser" || job.Project != "testproj" || @@ -223,12 +218,15 @@ func TestRestApi(t *testing.T) { job.Exclusive != 1 || job.MonitoringStatus != 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}}}) || job.StartTime.Unix() != 123456789 { 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 }); !ok { return diff --git a/init-db.go b/init-db.go index 6bafbf2..9ce3dfb 100644 --- a/init-db.go +++ b/init-db.go @@ -54,7 +54,8 @@ const JOBS_DB_SCHEMA string = ` CREATE TABLE tag ( id INTEGER PRIMARY KEY, 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 ( job_id INTEGER, diff --git a/repository/job.go b/repository/job.go index 44389be..ae42b5b 100644 --- a/repository/job.go +++ b/repository/job.go @@ -1,8 +1,6 @@ package repository import ( - "database/sql" - "github.com/ClusterCockpit/cc-backend/log" "github.com/ClusterCockpit/cc-backend/schema" sq "github.com/Masterminds/squirrel" @@ -52,17 +50,24 @@ func (r *JobRepository) FindById( return job, err } -func (r *JobRepository) Start(job schema.JobMeta) (res sql.Result, err error) { - res, err = r.DB.NamedExec(`INSERT INTO job ( +// 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) { + res, err := r.DB.NamedExec(`INSERT INTO job ( 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 ) VALUES ( :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 );`, 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( jobId int64, duration int32, @@ -91,30 +96,48 @@ func (r *JobRepository) Stop( } } - sql, args, err := stmt.ToSql() - - if err != nil { - log.Errorf("archiving job (dbid: %d) failed: %s", jobId, err.Error()) - } - - if _, err := r.DB.Exec(sql, args...); err != nil { + if _, err := stmt.RunWith(r.DB).Exec(); err != nil { 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 { _, err := r.DB.Exec(`INSERT INTO jobtag (job_id, tag_id) VALUES (?, ?)`, jobId, tagId) 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 if err := sq.Select("id").From("tag"). Where("tag.tag_type = ?", tagType).Where("tag.tag_name = ?", tagName). RunWith(r.DB).QueryRow().Scan(&tagId); err != nil { exists = false - return exists, tagId - } else { - return exists, tagId } + return }