mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2024-11-14 10:47:26 +01:00
fix: fixed and changed to footprint update by transactions
This commit is contained in:
parent
82e28f26d7
commit
63b9e619a4
@ -621,13 +621,12 @@ func (r *JobRepository) UpdateEnergy(
|
|||||||
}
|
}
|
||||||
|
|
||||||
var rawFootprint []byte
|
var rawFootprint []byte
|
||||||
|
|
||||||
if rawFootprint, err = json.Marshal(energyFootprint); err != nil {
|
if rawFootprint, err = json.Marshal(energyFootprint); err != nil {
|
||||||
log.Warnf("Error while marshaling energy footprint for job, DB ID '%v'", jobMeta.ID)
|
log.Warnf("Error while marshaling energy footprint for job INTO BYTES, DB ID '%v'", jobMeta.ID)
|
||||||
return stmt, err
|
return stmt, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return stmt.Set("energy_footprint", rawFootprint).Set("energy", (math.Round(totalEnergy*100) / 100)), nil
|
return stmt.Set("energy_footprint", string(rawFootprint)).Set("energy", (math.Round(totalEnergy*100) / 100)), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *JobRepository) UpdateFootprint(
|
func (r *JobRepository) UpdateFootprint(
|
||||||
@ -654,11 +653,10 @@ func (r *JobRepository) UpdateFootprint(
|
|||||||
}
|
}
|
||||||
|
|
||||||
var rawFootprint []byte
|
var rawFootprint []byte
|
||||||
|
|
||||||
if rawFootprint, err = json.Marshal(footprint); err != nil {
|
if rawFootprint, err = json.Marshal(footprint); err != nil {
|
||||||
log.Warnf("Error while marshaling footprint for job, DB ID '%v'", jobMeta.ID)
|
log.Warnf("Error while marshaling footprint for job INTO BYTES, DB ID '%v'", jobMeta.ID)
|
||||||
return stmt, err
|
return stmt, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return stmt.Set("footprint", rawFootprint), nil
|
return stmt.Set("footprint", string(rawFootprint)), nil
|
||||||
}
|
}
|
||||||
|
@ -49,7 +49,6 @@ func (r *JobRepository) TransactionEnd(t *Transaction) error {
|
|||||||
log.Warn("Error while committing SQL transactions")
|
log.Warn("Error while committing SQL transactions")
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -74,11 +73,16 @@ func (r *JobRepository) TransactionAddNamed(
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (r *JobRepository) TransactionAdd(t *Transaction, query string, args ...interface{}) (int64, error) {
|
func (r *JobRepository) TransactionAdd(t *Transaction, query string, args ...interface{}) (int64, error) {
|
||||||
res := t.tx.MustExec(query, args)
|
|
||||||
|
res, err := t.tx.Exec(query, args...)
|
||||||
|
if err != nil {
|
||||||
|
log.Errorf("TransactionAdd(), Exec() Error: %v", err)
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
|
||||||
id, err := res.LastInsertId()
|
id, err := res.LastInsertId()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("repository initDB(): %v", err)
|
log.Errorf("TransactionAdd(), LastInsertId() Error: %v", err)
|
||||||
return 0, err
|
return 0, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -24,12 +24,15 @@ func RegisterFootprintWorker() {
|
|||||||
gocron.NewTask(
|
gocron.NewTask(
|
||||||
func() {
|
func() {
|
||||||
s := time.Now()
|
s := time.Now()
|
||||||
log.Printf("Update Footprints started at %s using direct query execution", s.Format(time.RFC3339))
|
c := 0
|
||||||
|
ce := 0
|
||||||
|
cl := 0
|
||||||
|
log.Printf("Update Footprints started at %s", s.Format(time.RFC3339))
|
||||||
|
|
||||||
// t, err := jobRepo.TransactionInit()
|
t, err := jobRepo.TransactionInit()
|
||||||
// if err != nil {
|
if err != nil {
|
||||||
// log.Errorf("Failed TransactionInit %v", err)
|
log.Errorf("Failed TransactionInit %v", err)
|
||||||
// }
|
}
|
||||||
|
|
||||||
for _, cluster := range archive.Clusters {
|
for _, cluster := range archive.Clusters {
|
||||||
jobs, err := jobRepo.FindRunningJobs(cluster.Name)
|
jobs, err := jobRepo.FindRunningJobs(cluster.Name)
|
||||||
@ -47,10 +50,12 @@ func RegisterFootprintWorker() {
|
|||||||
scopes = append(scopes, schema.MetricScopeAccelerator)
|
scopes = append(scopes, schema.MetricScopeAccelerator)
|
||||||
|
|
||||||
for _, job := range jobs {
|
for _, job := range jobs {
|
||||||
// log.Debugf("Try job %d", job.JobID)
|
log.Debugf("Try job %d", job.JobID)
|
||||||
|
cl++
|
||||||
jobData, err := metricDataDispatcher.LoadData(job, allMetrics, scopes, context.Background(), 0) // 0 Resolution-Value retrieves highest res
|
jobData, err := metricDataDispatcher.LoadData(job, allMetrics, scopes, context.Background(), 0) // 0 Resolution-Value retrieves highest res
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("Error wile loading job data for footprint update: %v", err)
|
log.Errorf("Error wile loading job data for footprint update: %v", err)
|
||||||
|
ce++
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -65,6 +70,7 @@ func RegisterFootprintWorker() {
|
|||||||
nodeData, ok := data["node"]
|
nodeData, ok := data["node"]
|
||||||
if !ok {
|
if !ok {
|
||||||
// This should never happen ?
|
// This should never happen ?
|
||||||
|
ce++
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -92,33 +98,38 @@ func RegisterFootprintWorker() {
|
|||||||
stmt, err = jobRepo.UpdateFootprint(stmt, jobMeta)
|
stmt, err = jobRepo.UpdateFootprint(stmt, jobMeta)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("Update job (dbid: %d) failed at update Footprint step: %s", job.ID, err.Error())
|
log.Errorf("Update job (dbid: %d) failed at update Footprint step: %s", job.ID, err.Error())
|
||||||
|
ce++
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
stmt, err = jobRepo.UpdateEnergy(stmt, jobMeta)
|
stmt, err = jobRepo.UpdateEnergy(stmt, jobMeta)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("Update job (dbid: %d) failed at update Energy step: %s", job.ID, err.Error())
|
log.Errorf("Update job (dbid: %d) failed at update Energy step: %s", job.ID, err.Error())
|
||||||
|
ce++
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// Add WHERE Filter
|
// Add WHERE Filter
|
||||||
stmt = stmt.Where("job.id = ?", job.ID)
|
stmt = stmt.Where("job.id = ?", job.ID)
|
||||||
|
|
||||||
// query, args, err := stmt.ToSql()
|
query, args, err := stmt.ToSql()
|
||||||
// if err != nil {
|
if err != nil {
|
||||||
// log.Errorf("Failed in ToSQL conversion: %v", err)
|
log.Errorf("Failed in ToSQL conversion: %v", err)
|
||||||
// continue
|
ce++
|
||||||
// }
|
|
||||||
|
|
||||||
// jobRepo.TransactionAdd(t, query, args)
|
|
||||||
if err := jobRepo.Execute(stmt); err != nil {
|
|
||||||
log.Errorf("Update job footprint (dbid: %d) failed at db execute: %s", job.ID, err.Error())
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Args: JSON, JSON, ENERGY, JOBID
|
||||||
|
jobRepo.TransactionAdd(t, query, args...)
|
||||||
|
// if err := jobRepo.Execute(stmt); err != nil {
|
||||||
|
// log.Errorf("Update job footprint (dbid: %d) failed at db execute: %s", job.ID, err.Error())
|
||||||
|
// continue
|
||||||
|
// }
|
||||||
|
c++
|
||||||
log.Debugf("Finish Job %d", job.JobID)
|
log.Debugf("Finish Job %d", job.JobID)
|
||||||
}
|
}
|
||||||
|
jobRepo.TransactionCommit(t)
|
||||||
log.Debugf("Finish Cluster %s", cluster.Name)
|
log.Debugf("Finish Cluster %s", cluster.Name)
|
||||||
// jobRepo.TransactionCommit(t)
|
|
||||||
}
|
}
|
||||||
// jobRepo.TransactionEnd(t)
|
jobRepo.TransactionEnd(t)
|
||||||
log.Printf("Update Footprints is done and took %s", time.Since(s))
|
log.Printf("Updating %d (of %d; Skipped %d) Footprints is done and took %s", c, cl, ce, time.Since(s))
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user