Skip to content

Commit

Permalink
fix: fixed and changed to footprint update by transactions
Browse files Browse the repository at this point in the history
  • Loading branch information
spacehamster87 committed Oct 22, 2024
1 parent 82e28f2 commit 63b9e61
Show file tree
Hide file tree
Showing 3 changed files with 40 additions and 27 deletions.
10 changes: 4 additions & 6 deletions internal/repository/job.go
Original file line number Diff line number Diff line change
Expand Up @@ -621,13 +621,12 @@ func (r *JobRepository) UpdateEnergy(
}

var rawFootprint []byte

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.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(
Expand All @@ -654,11 +653,10 @@ func (r *JobRepository) UpdateFootprint(
}

var rawFootprint []byte

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.Set("footprint", rawFootprint), nil
return stmt.Set("footprint", string(rawFootprint)), nil
}
10 changes: 7 additions & 3 deletions internal/repository/transaction.go
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,6 @@ func (r *JobRepository) TransactionEnd(t *Transaction) error {
log.Warn("Error while committing SQL transactions")
return err
}

return nil
}

Expand All @@ -74,11 +73,16 @@ func (r *JobRepository) TransactionAddNamed(
}

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()
if err != nil {
log.Errorf("repository initDB(): %v", err)
log.Errorf("TransactionAdd(), LastInsertId() Error: %v", err)
return 0, err
}

Expand Down
47 changes: 29 additions & 18 deletions internal/taskManager/updateFootprintService.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,15 @@ func RegisterFootprintWorker() {
gocron.NewTask(
func() {
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()
// if err != nil {
// log.Errorf("Failed TransactionInit %v", err)
// }
t, err := jobRepo.TransactionInit()
if err != nil {
log.Errorf("Failed TransactionInit %v", err)
}

for _, cluster := range archive.Clusters {
jobs, err := jobRepo.FindRunningJobs(cluster.Name)
Expand All @@ -47,10 +50,12 @@ func RegisterFootprintWorker() {
scopes = append(scopes, schema.MetricScopeAccelerator)

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
if err != nil {
log.Errorf("Error wile loading job data for footprint update: %v", err)
ce++
continue
}

Expand All @@ -65,6 +70,7 @@ func RegisterFootprintWorker() {
nodeData, ok := data["node"]
if !ok {
// This should never happen ?
ce++
continue
}

Expand Down Expand Up @@ -92,33 +98,38 @@ func RegisterFootprintWorker() {
stmt, err = jobRepo.UpdateFootprint(stmt, jobMeta)
if err != nil {
log.Errorf("Update job (dbid: %d) failed at update Footprint step: %s", job.ID, err.Error())
ce++
continue
}
stmt, err = jobRepo.UpdateEnergy(stmt, jobMeta)
if err != nil {
log.Errorf("Update job (dbid: %d) failed at update Energy step: %s", job.ID, err.Error())
ce++
continue
}
// Add WHERE Filter
stmt = stmt.Where("job.id = ?", job.ID)

// query, args, err := stmt.ToSql()
// if err != nil {
// log.Errorf("Failed in ToSQL conversion: %v", err)
// continue
// }

// 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())
query, args, err := stmt.ToSql()
if err != nil {
log.Errorf("Failed in ToSQL conversion: %v", err)
ce++
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)
}
jobRepo.TransactionCommit(t)
log.Debugf("Finish Cluster %s", cluster.Name)
// jobRepo.TransactionCommit(t)
}
// jobRepo.TransactionEnd(t)
log.Printf("Update Footprints is done and took %s", time.Since(s))
jobRepo.TransactionEnd(t)
log.Printf("Updating %d (of %d; Skipped %d) Footprints is done and took %s", c, cl, ce, time.Since(s))
}))
}

0 comments on commit 63b9e61

Please sign in to comment.