Merge pull request #343 from ClusterCockpit/dev

Changes for Release v1.4.3

internal/archiver/archiver.go

@@ -60,12 +60,13 @@ func ArchiveJob(job *schema.Job, ctx context.Context) (*schema.JobMeta, error) {
 			max = math.Max(max, series.Statistics.Max)
 		}
 
+		// Round AVG Result to 2 Digits
 		jobMeta.Statistics[metric] = schema.JobStatistics{
 			Unit: schema.Unit{
 				Prefix: archive.GetMetricConfig(job.Cluster, metric).Unit.Prefix,
 				Base:   archive.GetMetricConfig(job.Cluster, metric).Unit.Base,
 			},
-			Avg: avg / float64(job.NumNodes),
+			Avg: (math.Round((avg/float64(job.NumNodes))*100) / 100),
 			Min: min,
 			Max: max,
 		}

internal/graph/schema.resolvers.go

@@ -303,6 +303,7 @@ func (r *queryResolver) JobMetrics(ctx context.Context, id string, metrics []str
 
 // JobsFootprints is the resolver for the jobsFootprints field.
 func (r *queryResolver) JobsFootprints(ctx context.Context, filter []*model.JobFilter, metrics []string) (*model.Footprints, error) {
+	// NOTE: Legacy Naming! This resolver is for normalized histograms in analysis view only - *Not* related to DB "footprint" column!
 	return r.jobsFootprints(ctx, filter, metrics)
 }
 

internal/metricDataDispatcher/dataLoader.go

@@ -170,6 +170,9 @@ func LoadData(job *schema.Job,
 			jd.AddNodeScope("mem_bw")
 		}
 
+		// Round Resulting Stat Values
+		jd.RoundMetricStats()
+
 		return jd, ttl, size
 	})
 

internal/metricdata/cc-metric-store.go

@@ -440,6 +440,23 @@ func (ccms *CCMetricStore) buildQueries(
 					continue
 				}
 
+				// Core -> Socket
+				if nativeScope == schema.MetricScopeCore && scope == schema.MetricScopeSocket {
+					sockets, _ := topology.GetSocketsFromCores(hwthreads)
+					for _, socket := range sockets {
+						queries = append(queries, ApiQuery{
+							Metric:     remoteName,
+							Hostname:   host.Hostname,
+							Aggregate:  true,
+							Type:       &coreString,
+							TypeIds:    intToStringSlice(topology.Socket[socket]),
+							Resolution: resolution,
+						})
+						assignedScope = append(assignedScope, scope)
+					}
+					continue
+				}
+
 				// Core -> Node
 				if nativeScope == schema.MetricScopeCore && scope == schema.MetricScopeNode {
 					cores, _ := topology.GetCoresFromHWThreads(hwthreads)
@@ -627,7 +644,7 @@ func (ccms *CCMetricStore) LoadNodeData(
 				req.Queries = append(req.Queries, ApiQuery{
 					Hostname:   node,
 					Metric:     ccms.toRemoteName(metric),
-					Resolution: 60, // Default for Node Queries
+					Resolution: 0, // Default for Node Queries: Will return metric $Timestep Resolution
 				})
 			}
 		}
@@ -1038,6 +1055,23 @@ func (ccms *CCMetricStore) buildNodeQueries(
 					continue
 				}
 
+				// Core -> Socket
+				if nativeScope == schema.MetricScopeCore && scope == schema.MetricScopeSocket {
+					sockets, _ := topology.GetSocketsFromCores(topology.Node)
+					for _, socket := range sockets {
+						queries = append(queries, ApiQuery{
+							Metric:     remoteName,
+							Hostname:   hostname,
+							Aggregate:  true,
+							Type:       &coreString,
+							TypeIds:    intToStringSlice(topology.Socket[socket]),
+							Resolution: resolution,
+						})
+						assignedScope = append(assignedScope, scope)
+					}
+					continue
+				}
+
 				// Core -> Node
 				if nativeScope == schema.MetricScopeCore && scope == schema.MetricScopeNode {
 					cores, _ := topology.GetCoresFromHWThreads(topology.Node)

internal/repository/job.go

@@ -217,11 +217,6 @@ func (r *JobRepository) UpdateMetadata(job *schema.Job, key, val string) (err er
 
 func (r *JobRepository) FetchFootprint(job *schema.Job) (map[string]float64, error) {
 	start := time.Now()
-	cachekey := fmt.Sprintf("footprint:%d", job.ID)
-	if cached := r.cache.Get(cachekey, nil); cached != nil {
-		job.Footprint = cached.(map[string]float64)
-		return job.Footprint, nil
-	}
 
 	if err := sq.Select("job.footprint").From("job").Where("job.id = ?", job.ID).
 		RunWith(r.stmtCache).QueryRow().Scan(&job.RawFootprint); err != nil {
@@ -238,7 +233,6 @@ func (r *JobRepository) FetchFootprint(job *schema.Job) (map[string]float64, err
 		return nil, err
 	}
 
-	r.cache.Put(cachekey, job.Footprint, len(job.Footprint), 24*time.Hour)
 	log.Debugf("Timer FetchFootprint %s", time.Since(start))
 	return job.Footprint, nil
 }
@@ -606,8 +600,11 @@ func (r *JobRepository) UpdateEnergy(
 				// FIXME: Needs sum as stats type
 			} else if sc.MetricConfig[i].Energy == "power" { // this metric has power as unit (Watt)
 				// Energy: Power (in Watts) * Time (in Seconds)
-				// Unit: ( W * s ) / 3600 / 1000 = kWh ; Rounded to 2 nearest digits
-				energy = math.Round(((LoadJobStat(jobMeta, fp, "avg")*float64(jobMeta.Duration))/3600/1000)*100) / 100
+				// Unit: (( W * s ) / 3600) / 1000 = kWh ; Rounded to 2 nearest digits: (Energy * 100) / 100
+				// Here: All-Node Metric Average * Number of Nodes * Job Runtime
+				// Note: Shared Jobs handled correctly since "Node Average" is based on partial resources, while "numNodes" factor is 1
+				metricNodeSum := LoadJobStat(jobMeta, fp, "avg") * float64(jobMeta.NumNodes) * float64(jobMeta.Duration)
+				energy = math.Round(((metricNodeSum/3600)/1000)*100) / 100
 			}
 		} else {
 			log.Warnf("Error while collecting energy metric %s for job, DB ID '%v', return '0.0'", fp, jobMeta.ID)

internal/taskManager/updateFootprintService.go

@@ -94,7 +94,7 @@ func RegisterFootprintWorker() {
 								}
 							}
 
-							// Add values rounded to 2 digits
+							// Add values rounded to 2 digits: repo.LoadStats may return unrounded
 							jobMeta.Statistics[metric] = schema.JobStatistics{
 								Unit: schema.Unit{
 									Prefix: archive.GetMetricConfig(job.Cluster, metric).Unit.Prefix,