Merge pull request #398 from ClusterCockpit/Refactor-job-struct

Refactor job struct

internal/repository/job.go

@@ -73,7 +73,7 @@ func scanJob(row interface{ Scan(...any) 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.StartTime, &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.RawFootprint, &job.Energy); err != nil {
 		log.Warnf("Error while scanning rows (Job): %v", err)
@@ -92,10 +92,9 @@ func scanJob(row interface{ Scan(...any) error }) (*schema.Job, error) {
 	}
 	job.RawFootprint = nil
 
-	job.StartTime = time.Unix(job.StartTimeUnix, 0)
 	// Always ensure accurate duration for running jobs
 	if job.State == schema.JobStateRunning {
-		job.Duration = int32(time.Since(job.StartTime).Seconds())
+		job.Duration = int32(time.Now().Unix() - job.StartTime)
 	}
 
 	return job, nil
@@ -582,7 +581,7 @@ func (r *JobRepository) MarkArchived(
 
 func (r *JobRepository) UpdateEnergy(
 	stmt sq.UpdateBuilder,
-	jobMeta *schema.JobMeta,
+	jobMeta *schema.Job,
 ) (sq.UpdateBuilder, error) {
 	/* Note: Only Called for Running Jobs during Intermediate Update or on Archiving */
 	sc, err := archive.GetSubCluster(jobMeta.Cluster, jobMeta.SubCluster)
@@ -632,7 +631,7 @@ func (r *JobRepository) UpdateEnergy(
 
 func (r *JobRepository) UpdateFootprint(
 	stmt sq.UpdateBuilder,
-	jobMeta *schema.JobMeta,
+	jobMeta *schema.Job,
 ) (sq.UpdateBuilder, error) {
 	/* Note: Only Called for Running Jobs during Intermediate Update or on Archiving */
 	sc, err := archive.GetSubCluster(jobMeta.Cluster, jobMeta.SubCluster)

internal/repository/jobCreate.go

@@ -29,7 +29,7 @@ const NamedJobInsert string = `INSERT INTO job (
   :exclusive, :monitoring_status, :smt, :job_state, :start_time, :duration, :walltime, :footprint,  :energy, :energy_footprint, :resources, :meta_data
 );`
 
-func (r *JobRepository) InsertJob(job *schema.JobMeta) (int64, error) {
+func (r *JobRepository) InsertJob(job *schema.Job) (int64, error) {
 	r.Mutex.Lock()
 	res, err := r.DB.NamedExec(NamedJobCacheInsert, job)
 	r.Mutex.Unlock()
@@ -87,7 +87,7 @@ func (r *JobRepository) SyncJobs() ([]*schema.Job, error) {
 
 // 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) {
+func (r *JobRepository) Start(job *schema.Job) (id int64, err error) {
 	job.RawFootprint, err = json.Marshal(job.Footprint)
 	if err != nil {
 		return -1, fmt.Errorf("REPOSITORY/JOB > encoding footprint field failed: %w", err)

internal/repository/jobFind.go

@@ -227,7 +227,7 @@ func (r *JobRepository) FindConcurrentJobs(
 	var startTime int64
 	var stopTime int64
 
-	startTime = job.StartTimeUnix
+	startTime = job.StartTime
 	hostname := job.Resources[0].Hostname
 
 	if job.State == schema.JobStateRunning {

internal/repository/job_test.go

@@ -24,7 +24,7 @@ func TestFind(t *testing.T) {
 
 	// fmt.Printf("%+v", job)
 
-	if job.ID != 5 {
+	if *job.ID != 5 {
 		t.Errorf("wrong summary for diagnostic 3\ngot: %d \nwant: 1366", job.JobID)
 	}
 }

internal/repository/stats.go

@@ -291,7 +291,7 @@ func (r *JobRepository) JobsStats(
 	return stats, nil
 }
 
-func LoadJobStat(job *schema.JobMeta, metric string, statType string) float64 {
+func LoadJobStat(job *schema.Job, metric string, statType string) float64 {
 	if stats, ok := job.Statistics[metric]; ok {
 		switch statType {
 		case "avg":
@@ -759,7 +759,6 @@ func (r *JobRepository) runningJobsMetricStatisticsHistogram(
 	filters []*model.JobFilter,
 	bins *int,
 ) []*model.MetricHistoPoints {
-
 	// Get Jobs
 	jobs, err := r.QueryJobs(ctx, filters, &model.PageRequest{Page: 1, ItemsPerPage: 500 + 1}, nil)
 	if err != nil {