mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2024-12-26 13:29:05 +01:00
Fix importers and add Energy footprint to import
This commit is contained in:
parent
210a7d3136
commit
cdd45ce88b
@ -8,6 +8,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"math"
|
||||||
"os"
|
"os"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
@ -84,7 +85,8 @@ func HandleImportFlag(flag string) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
name := fmt.Sprintf("%s_%s", fp, statType)
|
name := fmt.Sprintf("%s_%s", fp, statType)
|
||||||
job.Footprint[fp] = repository.LoadJobStat(&job, name, statType)
|
|
||||||
|
job.Footprint[name] = repository.LoadJobStat(&job, fp, statType)
|
||||||
}
|
}
|
||||||
|
|
||||||
job.RawFootprint, err = json.Marshal(job.Footprint)
|
job.RawFootprint, err = json.Marshal(job.Footprint)
|
||||||
@ -92,6 +94,34 @@ func HandleImportFlag(flag string) error {
|
|||||||
log.Warn("Error while marshaling job footprint")
|
log.Warn("Error while marshaling job footprint")
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
job.EnergyFootprint = make(map[string]float64)
|
||||||
|
var totalEnergy float64
|
||||||
|
var energy float64
|
||||||
|
|
||||||
|
for _, fp := range sc.EnergyFootprint {
|
||||||
|
if i, err := archive.MetricIndex(sc.MetricConfig, fp); err == nil {
|
||||||
|
// Note: For DB data, calculate and save as kWh
|
||||||
|
// Energy: Power (in Watts) * Time (in Seconds)
|
||||||
|
if sc.MetricConfig[i].Energy == "energy" { // this metric has energy as unit (Joules)
|
||||||
|
} else if sc.MetricConfig[i].Energy == "power" { // this metric has power as unit (Watt)
|
||||||
|
// Unit: ( W * s ) / 3600 / 1000 = kWh ; Rounded to 2 nearest digits
|
||||||
|
energy = math.Round(((repository.LoadJobStat(&job, fp, "avg")*float64(job.Duration))/3600/1000)*100) / 100
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
log.Warnf("Error while collecting energy metric %s for job, DB ID '%v', return '0.0'", fp, job.ID)
|
||||||
|
}
|
||||||
|
|
||||||
|
job.EnergyFootprint[fp] = energy
|
||||||
|
totalEnergy += energy
|
||||||
|
}
|
||||||
|
|
||||||
|
job.Energy = (math.Round(totalEnergy*100) / 100)
|
||||||
|
if job.RawEnergyFootprint, err = json.Marshal(job.EnergyFootprint); err != nil {
|
||||||
|
log.Warnf("Error while marshaling energy footprint for job INTO BYTES, DB ID '%v'", job.ID)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
job.RawResources, err = json.Marshal(job.Resources)
|
job.RawResources, err = json.Marshal(job.Resources)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn("Error while marshaling job resources")
|
log.Warn("Error while marshaling job resources")
|
||||||
|
@ -7,6 +7,7 @@ package importer
|
|||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"math"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@ -70,6 +71,7 @@ func InitDB() error {
|
|||||||
log.Errorf("cannot get subcluster: %s", err.Error())
|
log.Errorf("cannot get subcluster: %s", err.Error())
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
job.Footprint = make(map[string]float64)
|
job.Footprint = make(map[string]float64)
|
||||||
|
|
||||||
for _, fp := range sc.Footprint {
|
for _, fp := range sc.Footprint {
|
||||||
@ -90,6 +92,33 @@ func InitDB() error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
job.EnergyFootprint = make(map[string]float64)
|
||||||
|
var totalEnergy float64
|
||||||
|
var energy float64
|
||||||
|
|
||||||
|
for _, fp := range sc.EnergyFootprint {
|
||||||
|
if i, err := archive.MetricIndex(sc.MetricConfig, fp); err == nil {
|
||||||
|
// Note: For DB data, calculate and save as kWh
|
||||||
|
// Energy: Power (in Watts) * Time (in Seconds)
|
||||||
|
if sc.MetricConfig[i].Energy == "energy" { // this metric has energy as unit (Joules)
|
||||||
|
} else if sc.MetricConfig[i].Energy == "power" { // this metric has power as unit (Watt)
|
||||||
|
// Unit: ( W * s ) / 3600 / 1000 = kWh ; Rounded to 2 nearest digits
|
||||||
|
energy = math.Round(((repository.LoadJobStat(jobMeta, fp, "avg")*float64(jobMeta.Duration))/3600/1000)*100) / 100
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
log.Warnf("Error while collecting energy metric %s for job, DB ID '%v', return '0.0'", fp, jobMeta.ID)
|
||||||
|
}
|
||||||
|
|
||||||
|
job.EnergyFootprint[fp] = energy
|
||||||
|
totalEnergy += energy
|
||||||
|
}
|
||||||
|
|
||||||
|
job.Energy = (math.Round(totalEnergy*100) / 100)
|
||||||
|
if job.RawEnergyFootprint, err = json.Marshal(job.EnergyFootprint); err != nil {
|
||||||
|
log.Warnf("Error while marshaling energy footprint for job INTO BYTES, DB ID '%v'", jobMeta.ID)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
job.RawResources, err = json.Marshal(job.Resources)
|
job.RawResources, err = json.Marshal(job.Resources)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("repository initDB(): %v", err)
|
log.Errorf("repository initDB(): %v", err)
|
||||||
|
@ -15,10 +15,10 @@ import (
|
|||||||
|
|
||||||
const NamedJobInsert string = `INSERT INTO job (
|
const NamedJobInsert string = `INSERT INTO job (
|
||||||
job_id, user, project, cluster, subcluster, ` + "`partition`" + `, array_job_id, num_nodes, num_hwthreads, num_acc,
|
job_id, user, project, cluster, subcluster, ` + "`partition`" + `, array_job_id, num_nodes, num_hwthreads, num_acc,
|
||||||
exclusive, monitoring_status, smt, job_state, start_time, duration, walltime, footprint, resources, meta_data
|
exclusive, monitoring_status, smt, job_state, start_time, duration, walltime, footprint, energy, energy_footprint, resources, meta_data
|
||||||
) VALUES (
|
) VALUES (
|
||||||
:job_id, :user, :project, :cluster, :subcluster, :partition, :array_job_id, :num_nodes, :num_hwthreads, :num_acc,
|
:job_id, :user, :project, :cluster, :subcluster, :partition, :array_job_id, :num_nodes, :num_hwthreads, :num_acc,
|
||||||
:exclusive, :monitoring_status, :smt, :job_state, :start_time, :duration, :walltime, :footprint, :resources, :meta_data
|
: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.JobMeta) (int64, error) {
|
||||||
|
Loading…
Reference in New Issue
Block a user