Merge pull request #320 from ClusterCockpit/hotfix

Fixes for Bugfix Release 1.4.2

Author: moebiusband73

Makefile

@@ -2,7 +2,7 @@ TARGET = ./cc-backend
 VAR = ./var
 CFG = config.json .env
 FRONTEND = ./web/frontend
-VERSION = 1.4.1
+VERSION = 1.4.2
 GIT_HASH := $(shell git rev-parse --short HEAD || echo 'development')
 CURRENT_TIME = $(shell date +"%Y-%m-%d:T%H:%M:%S")
 LD_FLAGS = '-s -X main.date=${CURRENT_TIME} -X main.version=${VERSION} -X main.commit=${GIT_HASH}'

ReleaseNotes.md

@@ -1,4 +1,4 @@
-# `cc-backend` version 1.4.1
+# `cc-backend` version 1.4.2
 
 Supports job archive version 2 and database version 8.
 
@@ -12,7 +12,8 @@ For release specific notes visit the [ClusterCockpit Documentation](https://clus
   migration might require several hours!
 - You need to adapt the `cluster.json` configuration files in the job-archive,
   add new required attributes to the metric list and after that edit
-  `./job-archive/version.txt` to version 2.
+  `./job-archive/version.txt` to version 2. Only metrics that have the footprint
+  attribute set can be filtered and show up in the footprint UI and polar plot.
 - Continuous scrolling is default now in all job lists. You can change this back
   to paging globally, also every user can configure to use paging or continuous
   scrolling individually.

cmd/cc-backend/main.go

@@ -112,7 +112,7 @@ func main() {
 
 	if flagInit {
 		initEnv()
-		fmt.Print("Succesfully setup environment!\n")
+		fmt.Print("Successfully setup environment!\n")
 		fmt.Print("Please review config.json and .env and adjust it to your needs.\n")
 		fmt.Print("Add your job-archive at ./var/job-archive.\n")
 		os.Exit(0)

cmd/cc-backend/server.go

@@ -25,7 +25,6 @@ import (
 	"github.com/ClusterCockpit/cc-backend/internal/config"
 	"github.com/ClusterCockpit/cc-backend/internal/graph"
 	"github.com/ClusterCockpit/cc-backend/internal/graph/generated"
-	"github.com/ClusterCockpit/cc-backend/internal/repository"
 	"github.com/ClusterCockpit/cc-backend/internal/routerConfig"
 	"github.com/ClusterCockpit/cc-backend/pkg/log"
 	"github.com/ClusterCockpit/cc-backend/pkg/runtimeEnv"
@@ -314,9 +313,6 @@ func serverShutdown() {
 	// First shut down the server gracefully (waiting for all ongoing requests)
 	server.Shutdown(context.Background())
 
-	// Then, wait for any async jobStarts still pending...
-	repository.WaitForJobStart()
-
 	// Then, wait for any async archivings still pending...
 	archiver.WaitForArchiving()
 }

init/clustercockpit.service

@@ -1,5 +1,5 @@
 [Unit]
-Description=ClusterCockpit Web Server (Go edition)
+Description=ClusterCockpit Web Server
 Documentation=https://github.com/ClusterCockpit/cc-backend
 Wants=network-online.target
 After=network-online.target

internal/api/api_test.go

@@ -249,9 +249,6 @@ func TestRestApi(t *testing.T) {
 		if response.StatusCode != http.StatusCreated {
 			t.Fatal(response.Status, recorder.Body.String())
 		}
-
-		time.Sleep(1 * time.Second)
-
 		resolver := graph.GetResolverInstance()
 		job, err := restapi.JobRepository.Find(&TestJobId, &TestClusterName, &TestStartTime)
 		if err != nil {

internal/api/rest.go

@@ -123,18 +123,8 @@ func (api *RestApi) MountFrontendApiRoutes(r *mux.Router) {
 	}
 }
 
-// StartJobApiResponse model
-type StartJobApiResponse struct {
-	Message string `json:"msg"`
-}
-
-// DeleteJobApiResponse model
-type DeleteJobApiResponse struct {
-	Message string `json:"msg"`
-}
-
-// UpdateUserApiResponse model
-type UpdateUserApiResponse struct {
+// DefaultApiResponse model
+type DefaultJobApiResponse struct {
 	Message string `json:"msg"`
 }
 
@@ -341,7 +331,7 @@ func (api *RestApi) getJobs(rw http.ResponseWriter, r *http.Request) {
 	withMetadata := false
 	filter := &model.JobFilter{}
 	page := &model.PageRequest{ItemsPerPage: 25, Page: 1}
-	order := &model.OrderByInput{Field: "startTime", Order: model.SortDirectionEnumDesc}
+	order := &model.OrderByInput{Field: "startTime", Type: "col", Order: model.SortDirectionEnumDesc}
 
 	for key, vals := range r.URL.Query() {
 		switch key {
@@ -790,6 +780,11 @@ func (api *RestApi) startJob(rw http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// aquire lock to avoid race condition between API calls
+	var unlockOnce sync.Once
+	api.RepositoryMutex.Lock()
+	defer unlockOnce.Do(api.RepositoryMutex.Unlock)
+
 	// Check if combination of (job_id, cluster_id, start_time) already exists:
 	jobs, err := api.JobRepository.FindAll(&req.JobID, &req.Cluster, nil)
 	if err != nil && err != sql.ErrNoRows {
@@ -804,12 +799,27 @@ func (api *RestApi) startJob(rw http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	repository.TriggerJobStart(repository.JobWithUser{Job: &req, User: repository.GetUserFromContext(r.Context())})
+	id, err := api.JobRepository.Start(&req)
+	if err != nil {
+		handleError(fmt.Errorf("insert into database failed: %w", err), http.StatusInternalServerError, rw)
+		return
+	}
+	// unlock here, adding Tags can be async
+	unlockOnce.Do(api.RepositoryMutex.Unlock)
+
+	for _, tag := range req.Tags {
+		if _, err := api.JobRepository.AddTagOrCreate(repository.GetUserFromContext(r.Context()), id, tag.Type, tag.Name, tag.Scope); err != nil {
+			http.Error(rw, err.Error(), http.StatusInternalServerError)
+			handleError(fmt.Errorf("adding tag to new job %d failed: %w", id, err), http.StatusInternalServerError, rw)
+			return
+		}
+	}
 
+	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.WriteHeader(http.StatusCreated)
-	json.NewEncoder(rw).Encode(StartJobApiResponse{
-		Message: fmt.Sprintf("Successfully triggered job start"),
+	json.NewEncoder(rw).Encode(DefaultJobApiResponse{
+		Message: "success",
 	})
 }
 
@@ -892,7 +902,7 @@ func (api *RestApi) deleteJobById(rw http.ResponseWriter, r *http.Request) {
 	}
 	rw.Header().Add("Content-Type", "application/json")
 	rw.WriteHeader(http.StatusOK)
-	json.NewEncoder(rw).Encode(DeleteJobApiResponse{
+	json.NewEncoder(rw).Encode(DefaultJobApiResponse{
 		Message: fmt.Sprintf("Successfully deleted job %s", id),
 	})
 }
@@ -943,7 +953,7 @@ func (api *RestApi) deleteJobByRequest(rw http.ResponseWriter, r *http.Request)
 
 	rw.Header().Add("Content-Type", "application/json")
 	rw.WriteHeader(http.StatusOK)
-	json.NewEncoder(rw).Encode(DeleteJobApiResponse{
+	json.NewEncoder(rw).Encode(DefaultJobApiResponse{
 		Message: fmt.Sprintf("Successfully deleted job %d", job.ID),
 	})
 }
@@ -987,7 +997,7 @@ func (api *RestApi) deleteJobBefore(rw http.ResponseWriter, r *http.Request) {
 
 	rw.Header().Add("Content-Type", "application/json")
 	rw.WriteHeader(http.StatusOK)
-	json.NewEncoder(rw).Encode(DeleteJobApiResponse{
+	json.NewEncoder(rw).Encode(DefaultJobApiResponse{
 		Message: fmt.Sprintf("Successfully deleted %d jobs", cnt),
 	})
 }

internal/graph/schema.resolvers.go

@@ -82,8 +82,6 @@ func Connect(driver string, db string) {
 		if err != nil {
 			log.Fatal(err)
 		}
-
-		startJobStartWorker()
 	})
 }
 

internal/repository/dbConnection.go

@@ -79,12 +79,9 @@ func scanJob(row interface{ Scan(...interface{}) error }) (*schema.Job, error) {
 	}
 	job.RawFootprint = nil
 
-	// if err := json.Unmarshal(job.RawMetaData, &job.MetaData); err != nil {
-	// 	return nil, err
-	// }
-
 	job.StartTime = time.Unix(job.StartTimeUnix, 0)
-	if job.Duration == 0 && job.State == schema.JobStateRunning {
+	// Always ensure accurate duration for running jobs
+	if job.State == schema.JobStateRunning {
 		job.Duration = int32(time.Since(job.StartTime).Seconds())
 	}
 
@@ -457,6 +454,7 @@ func (r *JobRepository) AllocatedNodes(cluster string) (map[string]map[string]in
 	return subclusters, nil
 }
 
+// FIXME: Set duration to requested walltime?
 func (r *JobRepository) StopJobsExceedingWalltimeBy(seconds int) error {
 	start := time.Now()
 	res, err := sq.Update("job").

internal/repository/job.go

@@ -170,8 +170,7 @@ func BuildWhereClause(filter *model.JobFilter, query sq.SelectBuilder) sq.Select
 		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)
+		query = buildIntCondition("job.duration", filter.Duration, query)
 	}
 	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.

internal/repository/jobQuery.go

@@ -111,7 +111,7 @@ func BenchmarkDB_QueryJobs(b *testing.B) {
 	user := "mppi133h"
 	filter.User = &model.StringInput{Eq: &user}
 	page := &model.PageRequest{ItemsPerPage: 50, Page: 1}
-	order := &model.OrderByInput{Field: "startTime", Order: model.SortDirectionEnumDesc}
+	order := &model.OrderByInput{Field: "startTime", Type: "col", Order: model.SortDirectionEnumDesc}
 
 	b.Run("QueryJobs", func(b *testing.B) {
 		db := setup(b)

internal/repository/jobStartWorker.go

@@ -182,6 +182,7 @@ func setupTaglistRoute(i InfoType, r *http.Request) InfoType {
 	return i
 }
 
+// FIXME: Lots of redundant code. Needs refactoring
 func buildFilterPresets(query url.Values) map[string]interface{} {
 	filterPresets := map[string]interface{}{}