mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2026-08-31 00:47:15 +02:00
Merge branch 'feat/565-add-metric-tooltip' into metric-store-tickets
Resolve conflicts in the generated GraphQL code by regenerating it against
the merged schema. cc-lib v2.13.0 adds Tooltip to schema.MetricConfig and
schema.GlobalMetricListItem, so gqlgen now binds the tooltip field directly
and the hand-written globalMetricListItem/metricConfig resolvers introduced
on the tooltip branch are no longer needed.
Also migrate to the cc-lib v2.13.0 metric container types, which changed
from bare maps to structs carrying array-valued metric groups:
schema.JobData map -> {Metrics, Groups}
schema.ScopedJobStats map -> {Metrics, Groups}
job.Statistics map -> schema.JobStatisticsSet{Metrics, Groups}
Callers index .Metrics, return the zero struct instead of nil, and
deepCopy/DecodeJobStats now also carry the Groups payload through.
archive.GetStatistics returns the full JobStatisticsSet so group
statistics survive the round trip.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -299,7 +299,7 @@ func LoadAveragesFromArchive(
|
||||
}
|
||||
|
||||
for i, m := range metrics {
|
||||
if stat, ok := metaFile.Statistics[m]; ok {
|
||||
if stat, ok := metaFile.Statistics.Metrics[m]; ok {
|
||||
data[i] = append(data[i], schema.Float(stat.Avg))
|
||||
} else {
|
||||
data[i] = append(data[i], schema.NaN)
|
||||
@@ -323,7 +323,7 @@ func LoadStatsFromArchive(
|
||||
}
|
||||
|
||||
for _, m := range metrics {
|
||||
stat, ok := metaFile.Statistics[m]
|
||||
stat, ok := metaFile.Statistics.Metrics[m]
|
||||
if !ok {
|
||||
data[m] = schema.MetricStatistics{Min: 0.0, Avg: 0.0, Max: 0.0}
|
||||
continue
|
||||
@@ -349,19 +349,19 @@ func LoadScopedStatsFromArchive(
|
||||
data, err := ar.LoadJobStats(job)
|
||||
if err != nil {
|
||||
cclog.Errorf("Error while loading job stats from archiveBackend: %s", err.Error())
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
|
||||
return data, nil
|
||||
}
|
||||
|
||||
// GetStatistics returns all metric statistics for a job.
|
||||
// Returns a map of metric names to their job-level statistics.
|
||||
func GetStatistics(job *schema.Job) (map[string]schema.JobStatistics, error) {
|
||||
// GetStatistics returns all metric statistics for a job, including the
|
||||
// array-valued statistics groups (e.g. filesystems).
|
||||
func GetStatistics(job *schema.Job) (schema.JobStatisticsSet, error) {
|
||||
metaFile, err := ar.LoadJobMeta(job)
|
||||
if err != nil {
|
||||
cclog.Errorf("Error while loading job metadata from archiveBackend: %s", err.Error())
|
||||
return nil, err
|
||||
return schema.JobStatisticsSet{}, err
|
||||
}
|
||||
|
||||
return metaFile.Statistics, nil
|
||||
|
||||
@@ -63,8 +63,11 @@ func initClusterConfig() error {
|
||||
|
||||
if _, ok := metricLookup[mc.Name]; !ok {
|
||||
metricLookup[mc.Name] = schema.GlobalMetricListItem{
|
||||
Name: mc.Name, Scope: mc.Scope, Unit: mc.Unit, Footprint: mc.Footprint,
|
||||
Name: mc.Name, Scope: mc.Scope, Unit: mc.Unit, Footprint: mc.Footprint, Tooltip: mc.Tooltip,
|
||||
}
|
||||
} else if item := metricLookup[mc.Name]; item.Tooltip == "" && mc.Tooltip != "" {
|
||||
item.Tooltip = mc.Tooltip
|
||||
metricLookup[mc.Name] = item
|
||||
}
|
||||
|
||||
availability := schema.ClusterSupport{Cluster: cluster.Name}
|
||||
@@ -139,8 +142,10 @@ func initClusterConfig() error {
|
||||
userItem, ok := userMetricLookup[mc.Name]
|
||||
if !ok {
|
||||
userItem = schema.GlobalMetricListItem{
|
||||
Name: mc.Name, Scope: mc.Scope, Unit: mc.Unit, Footprint: mc.Footprint,
|
||||
Name: mc.Name, Scope: mc.Scope, Unit: mc.Unit, Footprint: mc.Footprint, Tooltip: mc.Tooltip,
|
||||
}
|
||||
} else if userItem.Tooltip == "" && mc.Tooltip != "" {
|
||||
userItem.Tooltip = mc.Tooltip
|
||||
}
|
||||
userItem.Availability = append(userItem.Availability, userAvailability)
|
||||
userMetricLookup[mc.Name] = userItem
|
||||
|
||||
@@ -93,7 +93,7 @@ func loadJobData(filename string, isCompressed bool) (schema.JobData, error) {
|
||||
f, err := os.Open(filename)
|
||||
if err != nil {
|
||||
cclog.Errorf("fsBackend LoadJobData()- %v", err)
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
@@ -101,7 +101,7 @@ func loadJobData(filename string, isCompressed bool) (schema.JobData, error) {
|
||||
r, err := gzip.NewReader(f)
|
||||
if err != nil {
|
||||
cclog.Errorf(" %v", err)
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
defer r.Close()
|
||||
|
||||
@@ -126,7 +126,7 @@ func loadJobStats(filename string, isCompressed bool) (schema.ScopedJobStats, er
|
||||
f, err := os.Open(filename)
|
||||
if err != nil {
|
||||
cclog.Errorf("fsBackend LoadJobStats()- %v", err)
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
@@ -134,13 +134,13 @@ func loadJobStats(filename string, isCompressed bool) (schema.ScopedJobStats, er
|
||||
r, err := gzip.NewReader(f)
|
||||
if err != nil {
|
||||
cclog.Errorf(" %v", err)
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
defer r.Close()
|
||||
|
||||
if config.Keys.Validate {
|
||||
if err := schema.Validate(schema.Data, r); err != nil {
|
||||
return nil, fmt.Errorf("validate job data: %v", err)
|
||||
return schema.ScopedJobStats{}, fmt.Errorf("validate job data: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,7 +148,7 @@ func loadJobStats(filename string, isCompressed bool) (schema.ScopedJobStats, er
|
||||
} else {
|
||||
if config.Keys.Validate {
|
||||
if err := schema.Validate(schema.Data, bufio.NewReader(f)); err != nil {
|
||||
return nil, fmt.Errorf("validate job data: %v", err)
|
||||
return schema.ScopedJobStats{}, fmt.Errorf("validate job data: %v", err)
|
||||
}
|
||||
}
|
||||
return DecodeJobStats(bufio.NewReader(f), filename)
|
||||
|
||||
@@ -129,7 +129,7 @@ func TestLoadJobData(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
for _, scopes := range data {
|
||||
for _, scopes := range data.Metrics {
|
||||
// fmt.Printf("Metric name: %s\n", name)
|
||||
|
||||
if _, exists := scopes[schema.MetricScopeNode]; !exists {
|
||||
|
||||
+53
-33
@@ -27,7 +27,7 @@ func DecodeJobData(r io.Reader, k string) (schema.JobData, error) {
|
||||
|
||||
if err, ok := data.(error); ok {
|
||||
cclog.Warn("Error in decoded job data set")
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
|
||||
return data.(schema.JobData), nil
|
||||
@@ -35,39 +35,59 @@ func DecodeJobData(r io.Reader, k string) (schema.JobData, error) {
|
||||
|
||||
func DecodeJobStats(r io.Reader, k string) (schema.ScopedJobStats, error) {
|
||||
jobData, err := DecodeJobData(r, k)
|
||||
// Convert schema.JobData to schema.ScopedJobStats
|
||||
if jobData != nil {
|
||||
scopedJobStats := make(schema.ScopedJobStats)
|
||||
for metric, metricData := range jobData {
|
||||
if _, ok := scopedJobStats[metric]; !ok {
|
||||
scopedJobStats[metric] = make(map[schema.MetricScope][]*schema.ScopedStats)
|
||||
}
|
||||
|
||||
for scope, jobMetric := range metricData {
|
||||
if _, ok := scopedJobStats[metric][scope]; !ok {
|
||||
scopedJobStats[metric][scope] = make([]*schema.ScopedStats, 0)
|
||||
}
|
||||
|
||||
for _, series := range jobMetric.Series {
|
||||
scopedJobStats[metric][scope] = append(scopedJobStats[metric][scope], &schema.ScopedStats{
|
||||
Hostname: series.Hostname,
|
||||
ID: series.ID,
|
||||
Data: &series.Statistics,
|
||||
})
|
||||
}
|
||||
|
||||
// So that one can later check len(scopedJobStats[metric][scope]): Remove from map if empty
|
||||
if len(scopedJobStats[metric][scope]) == 0 {
|
||||
delete(scopedJobStats[metric], scope)
|
||||
if len(scopedJobStats[metric]) == 0 {
|
||||
delete(scopedJobStats, metric)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return scopedJobStats, nil
|
||||
if err != nil {
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
return nil, err
|
||||
|
||||
// Convert schema.JobData to schema.ScopedJobStats
|
||||
scopedJobStats := schema.ScopedJobStats{
|
||||
Metrics: scopedStatsFromMetrics(jobData.Metrics),
|
||||
}
|
||||
|
||||
for _, group := range jobData.Groups {
|
||||
statsGroup := schema.ScopedStatsGroup{Key: group.Key}
|
||||
for _, inst := range group.Instances {
|
||||
statsGroup.Instances = append(statsGroup.Instances, schema.ScopedStatsGroupInstance{
|
||||
Name: inst.Name,
|
||||
Type: inst.Type,
|
||||
Metrics: scopedStatsFromMetrics(inst.Metrics),
|
||||
})
|
||||
}
|
||||
scopedJobStats.Groups = append(scopedJobStats.Groups, statsGroup)
|
||||
}
|
||||
|
||||
return scopedJobStats, nil
|
||||
}
|
||||
|
||||
// scopedStatsFromMetrics reduces the full time series of every metric/scope to
|
||||
// the per-series statistics. Scopes without any series are dropped so that
|
||||
// callers can rely on len(stats[metric][scope]) being non-zero when present.
|
||||
func scopedStatsFromMetrics(metrics map[string]schema.ScopedMetrics) map[string]schema.ScopedMetricStats {
|
||||
stats := make(map[string]schema.ScopedMetricStats, len(metrics))
|
||||
for metric, metricData := range metrics {
|
||||
scoped := make(schema.ScopedMetricStats, len(metricData))
|
||||
for scope, jobMetric := range metricData {
|
||||
if len(jobMetric.Series) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
series := make([]*schema.ScopedStats, 0, len(jobMetric.Series))
|
||||
for i := range jobMetric.Series {
|
||||
series = append(series, &schema.ScopedStats{
|
||||
Hostname: jobMetric.Series[i].Hostname,
|
||||
ID: jobMetric.Series[i].ID,
|
||||
Data: &jobMetric.Series[i].Statistics,
|
||||
})
|
||||
}
|
||||
scoped[scope] = series
|
||||
}
|
||||
|
||||
if len(scoped) > 0 {
|
||||
stats[metric] = scoped
|
||||
}
|
||||
}
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
func DecodeJobMeta(r io.Reader) (*schema.Job, error) {
|
||||
|
||||
@@ -23,7 +23,7 @@ func JobToParquetRow(meta *schema.Job, data *schema.JobData) (*ParquetJobRow, er
|
||||
}
|
||||
|
||||
var statisticsJSON []byte
|
||||
if meta.Statistics != nil {
|
||||
if len(meta.Statistics.Metrics) > 0 || len(meta.Statistics.Groups) > 0 {
|
||||
statisticsJSON, err = json.Marshal(meta.Statistics)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("marshal statistics: %w", err)
|
||||
|
||||
@@ -32,9 +32,9 @@ func TestParquetRowToJob(t *testing.T) {
|
||||
{Hostname: "node001", HWThreads: []int{0, 1, 2, 3}},
|
||||
{Hostname: "node002", HWThreads: []int{4, 5, 6, 7}},
|
||||
},
|
||||
Statistics: map[string]schema.JobStatistics{
|
||||
Statistics: schema.JobStatisticsSet{Metrics: map[string]schema.JobStatistics{
|
||||
"cpu_load": {Avg: 50.0, Min: 10.0, Max: 90.0},
|
||||
},
|
||||
}},
|
||||
Tags: []*schema.Tag{
|
||||
{Type: "test", Name: "tag1"},
|
||||
},
|
||||
@@ -49,7 +49,7 @@ func TestParquetRowToJob(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
data := &schema.JobData{
|
||||
data := &schema.JobData{Metrics: map[string]schema.ScopedMetrics{
|
||||
"cpu_load": {
|
||||
schema.MetricScopeNode: &schema.JobMetric{
|
||||
Unit: schema.Unit{Base: ""},
|
||||
@@ -62,7 +62,7 @@ func TestParquetRowToJob(t *testing.T) {
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}}
|
||||
|
||||
// Convert to parquet row
|
||||
row, err := JobToParquetRow(meta, data)
|
||||
@@ -134,10 +134,10 @@ func TestParquetRowToJob(t *testing.T) {
|
||||
t.Errorf("Resources[0].HWThreads len = %d, want 4", len(gotMeta.Resources[0].HWThreads))
|
||||
}
|
||||
|
||||
if len(gotMeta.Statistics) != 1 {
|
||||
t.Fatalf("Statistics len = %d, want 1", len(gotMeta.Statistics))
|
||||
if len(gotMeta.Statistics.Metrics) != 1 {
|
||||
t.Fatalf("Statistics len = %d, want 1", len(gotMeta.Statistics.Metrics))
|
||||
}
|
||||
if stat, ok := gotMeta.Statistics["cpu_load"]; !ok {
|
||||
if stat, ok := gotMeta.Statistics.Metrics["cpu_load"]; !ok {
|
||||
t.Error("Statistics missing cpu_load")
|
||||
} else if stat.Avg != 50.0 {
|
||||
t.Errorf("Statistics[cpu_load].Avg = %f, want 50.0", stat.Avg)
|
||||
@@ -163,7 +163,7 @@ func TestParquetRowToJob(t *testing.T) {
|
||||
if gotData == nil {
|
||||
t.Fatal("JobData is nil")
|
||||
}
|
||||
cpuLoad, ok := (*gotData)["cpu_load"]
|
||||
cpuLoad, ok := gotData.Metrics["cpu_load"]
|
||||
if !ok {
|
||||
t.Fatal("JobData missing cpu_load")
|
||||
}
|
||||
@@ -201,7 +201,7 @@ func TestParquetRowToJobNilOptionalFields(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
data := &schema.JobData{
|
||||
data := &schema.JobData{Metrics: map[string]schema.ScopedMetrics{
|
||||
"cpu_load": {
|
||||
schema.MetricScopeNode: &schema.JobMetric{
|
||||
Timestep: 60,
|
||||
@@ -210,7 +210,7 @@ func TestParquetRowToJobNilOptionalFields(t *testing.T) {
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}}
|
||||
|
||||
row, err := JobToParquetRow(meta, data)
|
||||
if err != nil {
|
||||
@@ -228,8 +228,8 @@ func TestParquetRowToJobNilOptionalFields(t *testing.T) {
|
||||
if gotMeta.Tags != nil {
|
||||
t.Errorf("Tags should be nil, got %v", gotMeta.Tags)
|
||||
}
|
||||
if gotMeta.Statistics != nil {
|
||||
t.Errorf("Statistics should be nil, got %v", gotMeta.Statistics)
|
||||
if len(gotMeta.Statistics.Metrics) != 0 || len(gotMeta.Statistics.Groups) != 0 {
|
||||
t.Errorf("Statistics should be empty, got %v", gotMeta.Statistics)
|
||||
}
|
||||
if gotMeta.MetaData != nil {
|
||||
t.Errorf("MetaData should be nil, got %v", gotMeta.MetaData)
|
||||
@@ -299,7 +299,7 @@ func TestRoundTripThroughParquetFile(t *testing.T) {
|
||||
if gotData == nil {
|
||||
t.Fatal("JobData is nil")
|
||||
}
|
||||
if _, ok := (*gotData)["cpu_load"]; !ok {
|
||||
if _, ok := gotData.Metrics["cpu_load"]; !ok {
|
||||
t.Error("JobData missing cpu_load")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -57,7 +57,7 @@ func makeTestJob(jobID int64) (*schema.Job, *schema.JobData) {
|
||||
},
|
||||
}
|
||||
|
||||
data := schema.JobData{
|
||||
data := schema.JobData{Metrics: map[string]schema.ScopedMetrics{
|
||||
"cpu_load": {
|
||||
schema.MetricScopeNode: &schema.JobMetric{
|
||||
Unit: schema.Unit{Base: ""},
|
||||
@@ -70,7 +70,7 @@ func makeTestJob(jobID int64) (*schema.Job, *schema.JobData) {
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}}
|
||||
|
||||
return meta, &data
|
||||
}
|
||||
@@ -132,7 +132,7 @@ func TestJobToParquetRowConversion(t *testing.T) {
|
||||
if err := json.Unmarshal(decompressed, &jobData); err != nil {
|
||||
t.Fatalf("unmarshal metric data: %v", err)
|
||||
}
|
||||
if _, ok := jobData["cpu_load"]; !ok {
|
||||
if _, ok := jobData.Metrics["cpu_load"]; !ok {
|
||||
t.Error("metric data missing cpu_load key")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -330,7 +330,7 @@ func (s3a *S3Archive) LoadJobData(job *schema.Job) (schema.JobData, error) {
|
||||
})
|
||||
if err != nil {
|
||||
cclog.Errorf("S3Archive LoadJobData() > GetObject error: %v", err)
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
defer result.Body.Close()
|
||||
|
||||
@@ -349,7 +349,7 @@ func (s3a *S3Archive) LoadJobData(job *schema.Job) (schema.JobData, error) {
|
||||
r, err := gzip.NewReader(result.Body)
|
||||
if err != nil {
|
||||
cclog.Errorf("S3Archive LoadJobData() > gzip error: %v", err)
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
defer r.Close()
|
||||
|
||||
@@ -381,14 +381,14 @@ func (s3a *S3Archive) LoadJobStats(job *schema.Job) (schema.ScopedJobStats, erro
|
||||
})
|
||||
if err != nil {
|
||||
cclog.Errorf("S3Archive LoadJobStats() > GetObject error: %v", err)
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
defer result.Body.Close()
|
||||
|
||||
if config.Keys.Validate {
|
||||
b, _ := io.ReadAll(result.Body)
|
||||
if err := schema.Validate(schema.Data, bytes.NewReader(b)); err != nil {
|
||||
return nil, fmt.Errorf("validate job data: %v", err)
|
||||
return schema.ScopedJobStats{}, fmt.Errorf("validate job data: %v", err)
|
||||
}
|
||||
return DecodeJobStats(bytes.NewReader(b), key)
|
||||
}
|
||||
@@ -400,14 +400,14 @@ func (s3a *S3Archive) LoadJobStats(job *schema.Job) (schema.ScopedJobStats, erro
|
||||
r, err := gzip.NewReader(result.Body)
|
||||
if err != nil {
|
||||
cclog.Errorf("S3Archive LoadJobStats() > gzip error: %v", err)
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
defer r.Close()
|
||||
|
||||
if config.Keys.Validate {
|
||||
b, _ := io.ReadAll(r)
|
||||
if err := schema.Validate(schema.Data, bytes.NewReader(b)); err != nil {
|
||||
return nil, fmt.Errorf("validate job data: %v", err)
|
||||
return schema.ScopedJobStats{}, fmt.Errorf("validate job data: %v", err)
|
||||
}
|
||||
return DecodeJobStats(bytes.NewReader(b), keyGz)
|
||||
}
|
||||
|
||||
@@ -252,7 +252,7 @@ func (sa *SqliteArchive) LoadJobData(job *schema.Job) (schema.JobData, error) {
|
||||
job.JobID, job.Cluster, job.StartTime).Scan(&dataBlob, &compressed)
|
||||
if err != nil {
|
||||
cclog.Errorf("SqliteArchive LoadJobData() > query error: %v", err)
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
key := fmt.Sprintf("%s:%d:%d", job.Cluster, job.JobID, job.StartTime)
|
||||
|
||||
@@ -261,7 +261,7 @@ func (sa *SqliteArchive) LoadJobData(job *schema.Job) (schema.JobData, error) {
|
||||
gzipReader, err := gzip.NewReader(reader)
|
||||
if err != nil {
|
||||
cclog.Errorf("SqliteArchive LoadJobData() > gzip error: %v", err)
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
defer gzipReader.Close()
|
||||
reader = gzipReader
|
||||
@@ -285,7 +285,7 @@ func (sa *SqliteArchive) LoadJobStats(job *schema.Job) (schema.ScopedJobStats, e
|
||||
job.JobID, job.Cluster, job.StartTime).Scan(&dataBlob, &compressed)
|
||||
if err != nil {
|
||||
cclog.Errorf("SqliteArchive LoadJobStats() > query error: %v", err)
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
key := fmt.Sprintf("%s:%d:%d", job.Cluster, job.JobID, job.StartTime)
|
||||
|
||||
@@ -294,7 +294,7 @@ func (sa *SqliteArchive) LoadJobStats(job *schema.Job) (schema.ScopedJobStats, e
|
||||
gzipReader, err := gzip.NewReader(reader)
|
||||
if err != nil {
|
||||
cclog.Errorf("SqliteArchive LoadJobStats() > gzip error: %v", err)
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
defer gzipReader.Close()
|
||||
reader = gzipReader
|
||||
@@ -303,7 +303,7 @@ func (sa *SqliteArchive) LoadJobStats(job *schema.Job) (schema.ScopedJobStats, e
|
||||
if config.Keys.Validate {
|
||||
data, _ := io.ReadAll(reader)
|
||||
if err := schema.Validate(schema.Data, bytes.NewReader(data)); err != nil {
|
||||
return nil, fmt.Errorf("validate job data: %v", err)
|
||||
return schema.ScopedJobStats{}, fmt.Errorf("validate job data: %v", err)
|
||||
}
|
||||
return DecodeJobStats(bytes.NewReader(data), key)
|
||||
}
|
||||
|
||||
+29
-29
@@ -89,7 +89,7 @@ func (ccms *InternalMetricStore) LoadData(
|
||||
queries, assignedScope, err := buildQueries(job, metrics, scopes, int64(resolution))
|
||||
if err != nil {
|
||||
cclog.Errorf("Error while building queries for jobId %d, Metrics %v, Scopes %v: %s", job.JobID, metrics, scopes, err.Error())
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
|
||||
// Verify assignment is correct - log any inconsistencies for debugging
|
||||
@@ -110,11 +110,11 @@ func (ccms *InternalMetricStore) LoadData(
|
||||
resBody, err := FetchData(req)
|
||||
if err != nil {
|
||||
cclog.Errorf("Error while fetching data : %s", err.Error())
|
||||
return nil, err
|
||||
return schema.JobData{}, err
|
||||
}
|
||||
|
||||
var errors []string
|
||||
jobData := make(schema.JobData)
|
||||
jobData := schema.JobData{Metrics: make(map[string]schema.ScopedMetrics)}
|
||||
|
||||
// Add safety check for potential index out of range errors
|
||||
if len(resBody.Results) != len(req.Queries) || len(assignedScope) != len(req.Queries) {
|
||||
@@ -139,8 +139,8 @@ func (ccms *InternalMetricStore) LoadData(
|
||||
continue
|
||||
}
|
||||
|
||||
if _, ok := jobData[metric]; !ok {
|
||||
jobData[metric] = make(map[schema.MetricScope]*schema.JobMetric)
|
||||
if _, ok := jobData.Metrics[metric]; !ok {
|
||||
jobData.Metrics[metric] = make(schema.ScopedMetrics)
|
||||
}
|
||||
|
||||
res := mc.Timestep
|
||||
@@ -148,14 +148,14 @@ func (ccms *InternalMetricStore) LoadData(
|
||||
res = int(row[0].Resolution)
|
||||
}
|
||||
|
||||
jobMetric, ok := jobData[metric][scope]
|
||||
jobMetric, ok := jobData.Metrics[metric][scope]
|
||||
if !ok {
|
||||
jobMetric = &schema.JobMetric{
|
||||
Unit: mc.Unit,
|
||||
Timestep: res,
|
||||
Series: make([]schema.Series, 0),
|
||||
}
|
||||
jobData[metric][scope] = jobMetric
|
||||
jobData.Metrics[metric][scope] = jobMetric
|
||||
}
|
||||
|
||||
for ndx, res := range row {
|
||||
@@ -181,11 +181,11 @@ func (ccms *InternalMetricStore) LoadData(
|
||||
})
|
||||
}
|
||||
|
||||
// So that one can later check len(jobData):
|
||||
// So that one can later check len(jobData.Metrics):
|
||||
if len(jobMetric.Series) == 0 {
|
||||
delete(jobData[metric], scope)
|
||||
if len(jobData[metric]) == 0 {
|
||||
delete(jobData, metric)
|
||||
delete(jobData.Metrics[metric], scope)
|
||||
if len(jobData.Metrics[metric]) == 0 {
|
||||
delete(jobData.Metrics, metric)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -407,7 +407,7 @@ func (ccms *InternalMetricStore) LoadScopedStats(
|
||||
queries, assignedScope, err := buildQueries(job, metrics, scopes, 0)
|
||||
if err != nil {
|
||||
cclog.Errorf("Error while building queries for jobId %d, Metrics %v, Scopes %v: %s", job.JobID, metrics, scopes, err.Error())
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
|
||||
req := APIQueryRequest{
|
||||
@@ -422,11 +422,11 @@ func (ccms *InternalMetricStore) LoadScopedStats(
|
||||
resBody, err := FetchData(req)
|
||||
if err != nil {
|
||||
cclog.Errorf("Error while fetching data : %s", err.Error())
|
||||
return nil, err
|
||||
return schema.ScopedJobStats{}, err
|
||||
}
|
||||
|
||||
var errors []string
|
||||
scopedJobStats := make(schema.ScopedJobStats)
|
||||
scopedJobStats := schema.ScopedJobStats{Metrics: make(map[string]schema.ScopedMetricStats)}
|
||||
|
||||
for i, row := range resBody.Results {
|
||||
if len(row) == 0 {
|
||||
@@ -437,12 +437,12 @@ func (ccms *InternalMetricStore) LoadScopedStats(
|
||||
metric := query.Metric
|
||||
scope := assignedScope[i]
|
||||
|
||||
if _, ok := scopedJobStats[metric]; !ok {
|
||||
scopedJobStats[metric] = make(map[schema.MetricScope][]*schema.ScopedStats)
|
||||
if _, ok := scopedJobStats.Metrics[metric]; !ok {
|
||||
scopedJobStats.Metrics[metric] = make(schema.ScopedMetricStats)
|
||||
}
|
||||
|
||||
if _, ok := scopedJobStats[metric][scope]; !ok {
|
||||
scopedJobStats[metric][scope] = make([]*schema.ScopedStats, 0)
|
||||
if _, ok := scopedJobStats.Metrics[metric][scope]; !ok {
|
||||
scopedJobStats.Metrics[metric][scope] = make([]*schema.ScopedStats, 0)
|
||||
}
|
||||
|
||||
for ndx, res := range row {
|
||||
@@ -456,7 +456,7 @@ func (ccms *InternalMetricStore) LoadScopedStats(
|
||||
|
||||
SanitizeStats(&res.Avg, &res.Min, &res.Max)
|
||||
|
||||
scopedJobStats[metric][scope] = append(scopedJobStats[metric][scope], &schema.ScopedStats{
|
||||
scopedJobStats.Metrics[metric][scope] = append(scopedJobStats.Metrics[metric][scope], &schema.ScopedStats{
|
||||
Hostname: query.Hostname,
|
||||
ID: id,
|
||||
Data: &schema.MetricStatistics{
|
||||
@@ -467,11 +467,11 @@ func (ccms *InternalMetricStore) LoadScopedStats(
|
||||
})
|
||||
}
|
||||
|
||||
// So that one can later check len(scopedJobStats[metric][scope]): Remove from map if empty
|
||||
if len(scopedJobStats[metric][scope]) == 0 {
|
||||
delete(scopedJobStats[metric], scope)
|
||||
if len(scopedJobStats[metric]) == 0 {
|
||||
delete(scopedJobStats, metric)
|
||||
// So that one can later check len(scopedJobStats.Metrics[metric][scope]): Remove from map if empty
|
||||
if len(scopedJobStats.Metrics[metric][scope]) == 0 {
|
||||
delete(scopedJobStats.Metrics[metric], scope)
|
||||
if len(scopedJobStats.Metrics[metric]) == 0 {
|
||||
delete(scopedJobStats.Metrics, metric)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -697,14 +697,14 @@ func (ccms *InternalMetricStore) LoadNodeListData(
|
||||
// Init Nested Map Data Structures If Not Found
|
||||
hostData, ok := data[query.Hostname]
|
||||
if !ok {
|
||||
hostData = make(schema.JobData)
|
||||
hostData = schema.JobData{Metrics: make(map[string]schema.ScopedMetrics)}
|
||||
data[query.Hostname] = hostData
|
||||
}
|
||||
|
||||
metricData, ok := hostData[metric]
|
||||
metricData, ok := hostData.Metrics[metric]
|
||||
if !ok {
|
||||
metricData = make(map[schema.MetricScope]*schema.JobMetric)
|
||||
data[query.Hostname][metric] = metricData
|
||||
metricData = make(schema.ScopedMetrics)
|
||||
hostData.Metrics[metric] = metricData
|
||||
}
|
||||
|
||||
scopeData, ok := metricData[scope]
|
||||
@@ -714,7 +714,7 @@ func (ccms *InternalMetricStore) LoadNodeListData(
|
||||
Timestep: res,
|
||||
Series: make([]schema.Series, 0),
|
||||
}
|
||||
data[query.Hostname][metric][scope] = scopeData
|
||||
metricData[scope] = scopeData
|
||||
}
|
||||
|
||||
for ndx, res := range row {
|
||||
|
||||
Reference in New Issue
Block a user