Merge branch 'dev' into add_detailed_nodelist

internal/auth/auth.go

@@ -143,19 +143,36 @@ func GetAuthInstance() *Authentication {
 	return authInstance
 }
 
-func persistUser(user *schema.User) {
+func handleTokenUser(tokenUser *schema.User) {
 	r := repository.GetUserRepository()
-	dbUser, err := r.GetUser(user.Username)
+	dbUser, err := r.GetUser(tokenUser.Username)
 
 	if err != nil && err != sql.ErrNoRows {
-		log.Errorf("Error while loading user '%s': %v", user.Username, err)
-	} else if err == sql.ErrNoRows { // Adds New User
-		if err := r.AddUser(user); err != nil {
-			log.Errorf("Error while adding user '%s' to DB: %v", user.Username, err)
+		log.Errorf("Error while loading user '%s': %v", tokenUser.Username, err)
+	} else if err == sql.ErrNoRows && config.Keys.JwtConfig.SyncUserOnLogin { // Adds New User
+		if err := r.AddUser(tokenUser); err != nil {
+			log.Errorf("Error while adding user '%s' to DB: %v", tokenUser.Username, err)
 		}
-	} else { // Update Existing
-		if err := r.UpdateUser(dbUser, user); err != nil {
-			log.Errorf("Error while updating user '%s' to DB: %v", user.Username, err)
+	} else if err == nil && config.Keys.JwtConfig.UpdateUserOnLogin { // Update Existing User
+		if err := r.UpdateUser(dbUser, tokenUser); err != nil {
+			log.Errorf("Error while updating user '%s' to DB: %v", dbUser.Username, err)
+		}
+	}
+}
+
+func handleOIDCUser(OIDCUser *schema.User) {
+	r := repository.GetUserRepository()
+	dbUser, err := r.GetUser(OIDCUser.Username)
+
+	if err != nil && err != sql.ErrNoRows {
+		log.Errorf("Error while loading user '%s': %v", OIDCUser.Username, err)
+	} else if err == sql.ErrNoRows && config.Keys.OpenIDConfig.SyncUserOnLogin { // Adds New User
+		if err := r.AddUser(OIDCUser); err != nil {
+			log.Errorf("Error while adding user '%s' to DB: %v", OIDCUser.Username, err)
+		}
+	} else if err == nil && config.Keys.OpenIDConfig.UpdateUserOnLogin { // Update Existing User
+		if err := r.UpdateUser(dbUser, OIDCUser); err != nil {
+			log.Errorf("Error while updating user '%s' to DB: %v", dbUser.Username, err)
 		}
 	}
 }

internal/auth/jwtCookieSession.go

@@ -198,8 +198,8 @@ func (ja *JWTCookieSessionAuthenticator) Login(
 			AuthSource: schema.AuthViaToken,
 		}
 
-		if jc.SyncUserOnLogin {
-			persistUser(user)
+		if jc.SyncUserOnLogin || jc.UpdateUserOnLogin {
+			handleTokenUser(user)
 		}
 	}
 

internal/auth/jwtSession.go

@@ -138,8 +138,8 @@ func (ja *JWTSessionAuthenticator) Login(
 			AuthSource: schema.AuthViaToken,
 		}
 
-		if config.Keys.JwtConfig.SyncUserOnLogin {
-			persistUser(user)
+		if config.Keys.JwtConfig.SyncUserOnLogin || config.Keys.JwtConfig.UpdateUserOnLogin {
+			handleTokenUser(user)
 		}
 	}
 

internal/auth/oidc.go

@@ -168,8 +168,8 @@ func (oa *OIDC) OAuth2Callback(rw http.ResponseWriter, r *http.Request) {
 		AuthSource: schema.AuthViaOIDC,
 	}
 
-	if config.Keys.OpenIDConfig.SyncUserOnLogin {
-		persistUser(user)
+	if config.Keys.OpenIDConfig.SyncUserOnLogin || config.Keys.OpenIDConfig.UpdateUserOnLogin {
+		handleOIDCUser(user)
 	}
 
 	oa.authentication.SaveSession(rw, r, user)

internal/repository/job.go

@@ -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(
@@ -643,7 +642,17 @@ func (r *JobRepository) UpdateFootprint(
 	footprint := make(map[string]float64)
 
 	for _, fp := range sc.Footprint {
-		statType := "avg"
+		var statType string
+		for _, gm := range archive.GlobalMetricList {
+			if gm.Name == fp {
+				statType = gm.Footprint
+			}
+		}
+
+		if statType != "avg" && statType != "min" && statType != "max" {
+			log.Warnf("unknown statType for footprint update: %s", statType)
+			return stmt, fmt.Errorf("unknown statType for footprint update: %s", statType)
+		}
 
 		if i, err := archive.MetricIndex(sc.MetricConfig, fp); err != nil {
 			statType = sc.MetricConfig[i].Footprint
@@ -654,11 +663,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
 }

internal/repository/transaction.go

@@ -49,7 +49,6 @@ func (r *JobRepository) TransactionEnd(t *Transaction) error {
 		log.Warn("Error while committing SQL transactions")
 		return err
 	}
-
 	return nil
 }
 
@@ -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
 	}
 

internal/taskManager/updateDurationService.go

@@ -7,14 +7,21 @@ package taskManager
 import (
 	"time"
 
+	"github.com/ClusterCockpit/cc-backend/internal/config"
 	"github.com/ClusterCockpit/cc-backend/pkg/log"
 	"github.com/go-co-op/gocron/v2"
 )
 
 func RegisterUpdateDurationWorker() {
-	log.Info("Register duration update service")
+	var frequency string
+	if config.Keys.CronFrequency.DurationWorker != "" {
+		frequency = config.Keys.CronFrequency.DurationWorker
+	} else {
+		frequency = "5m"
+	}
+	d, _ := time.ParseDuration(frequency)
+	log.Infof("Register Duration Update service with %s interval", frequency)
 
-	d, _ := time.ParseDuration("5m")
 	s.NewJob(gocron.DurationJob(d),
 		gocron.NewTask(
 			func() {

internal/taskManager/updateFootprintService.go

@@ -9,6 +9,7 @@ import (
 	"math"
 	"time"
 
+	"github.com/ClusterCockpit/cc-backend/internal/config"
 	"github.com/ClusterCockpit/cc-backend/internal/metricDataDispatcher"
 	"github.com/ClusterCockpit/cc-backend/pkg/archive"
 	"github.com/ClusterCockpit/cc-backend/pkg/log"
@@ -18,18 +19,28 @@ import (
 )
 
 func RegisterFootprintWorker() {
-	log.Info("Register Footprint Update service")
-	d, _ := time.ParseDuration("10m")
+	var frequency string
+	if config.Keys.CronFrequency.FootprintWorker != "" {
+		frequency = config.Keys.CronFrequency.FootprintWorker
+	} else {
+		frequency = "10m"
+	}
+	d, _ := time.ParseDuration(frequency)
+	log.Infof("Register Footprint Update service with %s interval", frequency)
+
 	s.NewJob(gocron.DurationJob(d),
 		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)
@@ -47,10 +58,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
 						}
 
@@ -65,6 +78,7 @@ func RegisterFootprintWorker() {
 							nodeData, ok := data["node"]
 							if !ok {
 								// This should never happen ?
+								ce++
 								continue
 							}
 
@@ -92,33 +106,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))
 			}))
 }