ClusterCockpit/cc-backend
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-backend synced 2024-11-10 08:57:25 +01:00
Code Issues Packages Releases Activity

Update cc-metric-store client; Bugfix in LoadNodeData

Browse Source
This commit is contained in:
Lou Knauer 2022-02-02 13:04:38 +01:00
parent 3d5aa9f904
commit f7959cb1bd
1 changed files with 35 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
metricdata/cc-metric-store.go
View File

@ -6,6 +6,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log"
"net/http" "net/http"
"time" "time"
@ -42,6 +43,11 @@ type ApiQuery struct {
SubTypeIds []int `json:"subtype-ids,omitempty"` SubTypeIds []int `json:"subtype-ids,omitempty"`
} }
type ApiQueryResponse struct {
Queries []ApiQuery `json:"queries,omitempty"`
Results [][]ApiMetricData `json:"results"`
}
type ApiMetricData struct { type ApiMetricData struct {
Error *string `json:"error"` Error *string `json:"error"`
From int64 `json:"from"` From int64 `json:"from"`
@ -90,7 +96,7 @@ func (ccms *CCMetricStore) toLocalName(metric string) string {
return metric return metric
} }
func (ccms *CCMetricStore) doRequest(ctx context.Context, body *ApiQueryRequest) ([][]ApiMetricData, error) { func (ccms *CCMetricStore) doRequest(ctx context.Context, body *ApiQueryRequest) (*ApiQueryResponse, error) {
buf := &bytes.Buffer{} buf := &bytes.Buffer{}
if err := json.NewEncoder(buf).Encode(body); err != nil { if err := json.NewEncoder(buf).Encode(body); err != nil {
return nil, err return nil, err
@ -113,12 +119,12 @@ func (ccms *CCMetricStore) doRequest(ctx context.Context, body *ApiQueryRequest)
return nil, fmt.Errorf("'%s': HTTP Status: %s", ccms.queryEndpoint, res.Status) return nil, fmt.Errorf("'%s': HTTP Status: %s", ccms.queryEndpoint, res.Status)
} }
var resBody [][]ApiMetricData var resBody ApiQueryResponse
if err := json.NewDecoder(bufio.NewReader(res.Body)).Decode(&resBody); err != nil { if err := json.NewDecoder(bufio.NewReader(res.Body)).Decode(&resBody); err != nil {
return nil, err return nil, err
} }
return resBody, nil return &resBody, nil
} }
func (ccms *CCMetricStore) LoadData(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context) (schema.JobData, error) { func (ccms *CCMetricStore) LoadData(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context) (schema.JobData, error) {
@ -142,7 +148,7 @@ func (ccms *CCMetricStore) LoadData(job *schema.Job, metrics []string, scopes []
} }
var jobData schema.JobData = make(schema.JobData) var jobData schema.JobData = make(schema.JobData)
for i, row := range resBody { for i, row := range resBody.Results {
query := req.Queries[i] query := req.Queries[i]
metric := ccms.toLocalName(query.Metric) metric := ccms.toLocalName(query.Metric)
scope := assignedScope[i] scope := assignedScope[i]
@ -421,7 +427,7 @@ func (ccms *CCMetricStore) LoadStats(job *schema.Job, metrics []string, ctx cont
} }
stats := make(map[string]map[string]schema.MetricStatistics, len(metrics)) stats := make(map[string]map[string]schema.MetricStatistics, len(metrics))
for i, res := range resBody { for i, res := range resBody.Results {
query := req.Queries[i] query := req.Queries[i]
metric := ccms.toLocalName(query.Metric) metric := ccms.toLocalName(query.Metric)
data := res[0] data := res[0]
@ -455,12 +461,15 @@ func (ccms *CCMetricStore) LoadNodeData(cluster, partition string, metrics, node
Cluster: cluster, Cluster: cluster,
From: from.Unix(), From: from.Unix(),
To: to.Unix(), To: to.Unix(),
WithStats: false, WithStats: true,
WithData: true, WithData: true,
} }
if nodes == nil { if nodes == nil {
req.ForAllNodes = metrics for _, metric := range metrics {
log.Printf("local-name: %s, remote-name: %s, renamings: %#v", metric, ccms.toRemoteName(metric), ccms.here2there)
req.ForAllNodes = append(req.ForAllNodes, ccms.toRemoteName(metric))
}
} else { } else {
for _, node := range nodes { for _, node := range nodes {
for _, metric := range metrics { for _, metric := range metrics {
@ -479,14 +488,26 @@ func (ccms *CCMetricStore) LoadNodeData(cluster, partition string, metrics, node
_ = resBody _ = resBody
data := make(map[string]map[string][]*schema.JobMetric) data := make(map[string]map[string][]*schema.JobMetric)
for i, res := range resBody { for i, res := range resBody.Results {
query := req.Queries[i] var query ApiQuery
if resBody.Queries != nil {
query = resBody.Queries[i]
} else {
query = req.Queries[i]
}
log.Printf("results for %#v: %#v\n", query, res)
metric := ccms.toLocalName(query.Metric) metric := ccms.toLocalName(query.Metric)
qdata := res[0] qdata := res[0]
if qdata.Error != nil { if qdata.Error != nil {
return nil, fmt.Errorf("fetching %s for node %s failed: %s", query.Metric, query.Hostname, *qdata.Error) return nil, fmt.Errorf("fetching %s for node %s failed: %s", query.Metric, query.Hostname, *qdata.Error)
} }
if qdata.Avg.IsNaN() || qdata.Min.IsNaN() || qdata.Max.IsNaN() {
return nil, fmt.Errorf("fetching %s for node %s failed: %s", metric, query.Hostname, "avg/min/max is NaN")
}
hostdata, ok := data[query.Hostname] hostdata, ok := data[query.Hostname]
if !ok { if !ok {
hostdata = make(map[string][]*schema.JobMetric) hostdata = make(map[string][]*schema.JobMetric)
@ -502,6 +523,11 @@ func (ccms *CCMetricStore) LoadNodeData(cluster, partition string, metrics, node
{ {
Hostname: query.Hostname, Hostname: query.Hostname,
Data: qdata.Data, Data: qdata.Data,
Statistics: &schema.MetricStatistics{
Avg: float64(qdata.Avg),
Min: float64(qdata.Min),
Max: float64(qdata.Max),
},
}, },
}, },
}) })