From d36bd886d5ec8c90e367bbb5c5bc2bfe52629b4c Mon Sep 17 00:00:00 2001 From: sgozel Date: Mon, 12 Oct 2026 17:06:55 +0200 Subject: [PATCH 1/6] Add VM endpoint and DCGM metrics --- config.yaml.sample | 3 + elastic/elastic.go | 4 +- examples/gpu_temperature.py | 3 +- examples/vm_check_gpu_temperature.py | 81 +++++++++++ examples/vm_gpu_temperature.py | 43 ++++++ handler/vm_dcgm.go | 68 +++++++++ hpcdata.go | 7 + util/config.go | 27 +++- victoriametrics/victoriametrics.go | 197 +++++++++++++++++++++++++++ 9 files changed, 422 insertions(+), 11 deletions(-) create mode 100644 examples/vm_check_gpu_temperature.py create mode 100644 examples/vm_gpu_temperature.py create mode 100644 handler/vm_dcgm.go create mode 100644 victoriametrics/victoriametrics.go diff --git a/config.yaml.sample b/config.yaml.sample index 08e329b..7f65a4d 100644 --- a/config.yaml.sample +++ b/config.yaml.sample @@ -6,6 +6,9 @@ elastic: url: 'https://examle.com:8080' username: 'my-username' password: 'my-password' +victoriametrics: + url: 'http://example.com:8400' + token: 'my-vm-token' security: # if a group/username appears in the list below, then it is allowed to query data for any job, # even if it would otherwise not be accessible by the authorized user diff --git a/elastic/elastic.go b/elastic/elastic.go index 1aa2779..eabcb50 100644 --- a/elastic/elastic.go +++ b/elastic/elastic.go @@ -536,8 +536,8 @@ func (c *Client) GetChassisPower(nodes []util.Node, from time.Time, to time.Time } type DcgmDataIndexed struct { - GpuIndex int - Data []float64 + GpuIndex int `json:"gpu_id"` + Data []float64 `json:"data"` } type DcgmMetric struct { Time []time.Time diff --git a/examples/gpu_temperature.py b/examples/gpu_temperature.py index e370954..a22258c 100644 --- a/examples/gpu_temperature.py +++ b/examples/gpu_temperature.py @@ -3,10 +3,9 @@ import yaml import matplotlib.pyplot as plt -import matplotlib.dates as mdates import requests -from common import Config, generate_token, make_format +from common import Config, generate_token if __name__ == '__main__': with open(os.path.join(os.path.dirname(__file__), 'config.yaml')) as f: diff --git a/examples/vm_check_gpu_temperature.py b/examples/vm_check_gpu_temperature.py new file mode 100644 index 0000000..7b7edac --- /dev/null +++ b/examples/vm_check_gpu_temperature.py @@ -0,0 +1,81 @@ +import datetime +import os +import yaml + +import matplotlib.pyplot as plt +import requests + +from common import Config, generate_token + + +def fetch(url: str, headers: dict): + r = requests.get(url, headers=headers) + r.raise_for_status() + return r.json() + + +if __name__ == '__main__': + with open(os.path.join(os.path.dirname(__file__), 'config.yaml')) as f: + config: Config = yaml.safe_load(f) + jobid = config['jobid'] + cluster = config['cluster'] + metric = 'gpu_temp' + token = generate_token(config) + auth_header = {'Authorization': f'Bearer {token}'} + + es_url = f"{config['base_url']}/metrics/{cluster}/{jobid}/gpu/temperature" + vm_url = f"{config['base_url']}/metrics/{cluster}/{jobid}/vm/gpu/{metric}" + + es_data = fetch(es_url, auth_header) + vm_data = fetch(vm_url, auth_header) + + es_time = [datetime.datetime.fromtimestamp(x) for x in es_data['time']] + vm_time = [datetime.datetime.fromtimestamp(x) for x in vm_data['time']] + + all_nodes = set(es_data['nodes'].keys()) | set(vm_data['nodes'].keys()) + + cols = ['r', 'b', 'g', 'm'] + + for nodeid in sorted(all_nodes): + # ES returns a list of objects per GPU + es_gpus = {g['gpu_id']: g for g in es_data['nodes'].get(nodeid, [])} + # VM returns a dict with a metric key containing a list of objects per GPU + vm_node = vm_data['nodes'].get(nodeid, {}) + vm_gpus = {g['gpu_id']: g for g in vm_node.get(metric, [])} + + fig, ax = plt.subplots() + fig.set_figwidth(19) + fig.set_figheight(10) + + all_gpu_ids = set(es_gpus.keys()) | set(vm_gpus.keys()) + for gpu_id in sorted(all_gpu_ids): + if gpu_id in es_gpus: + g = es_gpus[gpu_id] + ax.plot(es_time, g['temperature'], + color=cols[gpu_id], + label=f'ES GPU {gpu_id}') + if gpu_id in vm_gpus: + g = vm_gpus[gpu_id] + ax.plot(vm_time, g['data'], + color=cols[gpu_id], + linestyle='--', + label=f'VM GPU {gpu_id}') + + if gpu_id in es_gpus and gpu_id in vm_gpus: + es_mean = sum(es_gpus[gpu_id]['temperature']) / len(es_gpus[gpu_id]['temperature']) + vm_mean = sum(vm_gpus[gpu_id]['data']) / len(vm_gpus[gpu_id]['data']) + print( + f'{nodeid} GPU {gpu_id}: ' + f'ES points={len(es_gpus[gpu_id]["temperature"])}, ' + f'VM points={len(vm_gpus[gpu_id]["data"])}, ' + f'ES mean={es_mean:.2f}°C, ' + f'VM mean={vm_mean:.2f}°C' + ) + + ax.set_title(f'GPU temperature comparison for {nodeid}') + ax.set_xlabel('Time') + ax.set_ylabel('Temperature (°C)') + ax.grid(True) + fig.legend() + fig.tight_layout() + plt.show() diff --git a/examples/vm_gpu_temperature.py b/examples/vm_gpu_temperature.py new file mode 100644 index 0000000..a3f2845 --- /dev/null +++ b/examples/vm_gpu_temperature.py @@ -0,0 +1,43 @@ +import datetime +import os +import yaml + +import matplotlib.pyplot as plt +import requests + +from common import Config, generate_token + +if __name__ == '__main__': + with open(os.path.join(os.path.dirname(__file__), 'config.yaml')) as f: + config: Config = yaml.safe_load(f) + jobid = config['jobid'] + cluster = config['cluster'] + metric = 'gpu_temp' + token = generate_token(config) + auth_header = {'Authorization': f'Bearer {token}'} + r = requests.get(f"{config['base_url']}/metrics/{cluster}/{jobid}/vm/gpu/{metric}", headers=auth_header) + r.raise_for_status() + + response = r.json() + time = [datetime.datetime.fromtimestamp(x) for x in response['time']] + nodes = response['nodes'] + + fig, ax = plt.subplots() + fig.set_figwidth(19) + fig.set_figheight(10) + + for nodeid, nodeData in nodes.items(): + for gpuData in nodeData[metric]: + ax.plot(time, gpuData['data'], + marker='.', + label=f'GPU {gpuData["gpu_id"]}') + # only plot GPU temperatures of the first node + break + + ax.set_title(f'VictoriaMetrics {metric}') + ax.set_xlabel('Time') + ax.set_ylabel('Temperature (°C)') + ax.grid(True) + fig.legend() + fig.tight_layout() + plt.show() diff --git a/handler/vm_dcgm.go b/handler/vm_dcgm.go new file mode 100644 index 0000000..3a0c358 --- /dev/null +++ b/handler/vm_dcgm.go @@ -0,0 +1,68 @@ +package handler + +import ( + "encoding/json" + "fmt" + "net/http" + "slices" + + "github.com/gorilla/mux" + + "cscs.ch/hpcdata/elastic" + "cscs.ch/hpcdata/logging" + "cscs.ch/hpcdata/util" + "cscs.ch/hpcdata/victoriametrics" +) + +type vmDcgm struct { + config *util.Config + esclient *elastic.Client + vmclient *victoriametrics.Client +} + +func GetVMDcgmHandler(config *util.Config, esclient *elastic.Client, vmclient *victoriametrics.Client) func(w http.ResponseWriter, r *http.Request) { + return wrap(vmDcgm{config, esclient, vmclient}) +} + +func (h vmDcgm) Get(w http.ResponseWriter, r *http.Request) { + logger := logging.GetReqLogger(r) + job, from, to := panic_if_no_access(r, h.esclient, h.config) + + vars := mux.Vars(r) + systemName := vars["system_name"] + metricName := vars["metric_name"] + + info, ok := victoriametrics.SupportedMetrics[metricName] + if !ok { + pie(logger.Warn, condition_error{fmt.Sprintf("Unsupported VM DCGM metric: %v", metricName)}, "", http.StatusBadRequest) + } + + logger.Debug().Msgf("Passed all security checks to fetch VM DCGM metric=%v for job=%+v in the time window from=%v to=%v", metricName, job, from, to) + + cluster_config, err := h.config.GetClusterConfig(systemName) + pie(logger.Warn, err, "", http.StatusBadRequest) + + // [TO DO] 8 lines below are taken from dcgm.go ---> refactor to avoid duplicate + nodes := job.Nodes + if node_id, exists := vars["node_id"]; exists { + nodes = []util.Node{{Nid: node_id}} + // security check that the node is part of the job + if !slices.ContainsFunc(job.Nodes, func(n util.Node) bool { return n.Nid == node_id }) { + pie(logger.Warn, condition_error{"The requested node_id is not part of the job"}, "", http.StatusBadRequest) + } + } + + dcgmData, err := h.vmclient.GetDcgmMetric(metricName, nodes, from, to, cluster_config.Name, logger) + pie(logger.Error, err, fmt.Sprintf("Failed getting VM DCGM metric %v", metricName), http.StatusInternalServerError) + + ret := struct { + Time []epochTime `json:"time"` + Nodes map[string]map[string]any `json:"nodes"` + }{as_epoch_array(dcgmData.Time), map[string]map[string]any{}} + for nid, dcgmMetric := range dcgmData.MetricByNode { + ret.Nodes[nid] = map[string]any{metricName: dcgmMetric, fmt.Sprintf("%v_unit", metricName): info.Unit} + } + + write_bytes, err := json.Marshal(ret) + _, _ = w.Write(write_bytes) +} diff --git a/hpcdata.go b/hpcdata.go index 278afaf..401fd7b 100644 --- a/hpcdata.go +++ b/hpcdata.go @@ -14,6 +14,7 @@ import ( "cscs.ch/hpcdata/handler" "cscs.ch/hpcdata/logging" "cscs.ch/hpcdata/util" + "cscs.ch/hpcdata/victoriametrics" ) type LimitBodyMiddleware struct { @@ -40,6 +41,8 @@ func main() { esclient := elastic.NewClient(config) + vmclient := victoriametrics.NewClient(config) + handler.InitRedis(config.RedisConfig) for _, jwt_signing_url := range config.OauthSigners { @@ -53,6 +56,10 @@ func main() { reqHandler.HandleFunc("/metrics/{system_name}/{job_id}/gpu/temperature", handler.GetGpuTemperatureHandler(config, esclient)) reqHandler.HandleFunc("/metrics/{system_name}/{job_id}/{node_id}/gpu/temperature", handler.GetGpuTemperatureHandler(config, esclient)) + // VictoriaMetrics DCGM metrics + reqHandler.HandleFunc("/metrics/{system_name}/{job_id}/vm/gpu/{metric_name}", handler.GetVMDcgmHandler(config, esclient, vmclient)) + reqHandler.HandleFunc("/metrics/{system_name}/{job_id}/{node_id}/vm/gpu/{metric_name}", handler.GetVMDcgmHandler(config, esclient, vmclient)) + // DCGM data reqHandler.HandleFunc("/metrics/{system_name}/{job_id}/gpu/utilization", handler.GetDcgmData(config, esclient, "gpu_utilization")) reqHandler.HandleFunc("/metrics/{system_name}/{job_id}/{node_id}/gpu/utilization", handler.GetDcgmData(config, esclient, "gpu_utilization")) diff --git a/util/config.go b/util/config.go index 08ddf14..ea8db9a 100644 --- a/util/config.go +++ b/util/config.go @@ -28,6 +28,10 @@ type ElasticConfig struct { Username string `yaml:"username"` Password string `yaml:"password"` } +type VictoriaMetricsConfig struct { + URL string `yaml:"url"` + Token string `yaml:"token"` +} type SecurityConfig struct { AllowAnyJob []string `yaml:"allow_any_job"` } @@ -41,13 +45,14 @@ type RedisConfig struct { Password string `yaml:"password"` } type Config struct { - Server ServerConfig `yaml:"server"` - Database DatabaseConfig `yaml:"db"` - Elastic ElasticConfig `yaml:"elastic"` - OauthSigners []string `yaml:"openid"` - Clusters []ClusterConfig `yaml:"clusters"` - Security SecurityConfig `yaml:"security"` - RedisConfig RedisConfig `yaml:"redis"` + Server ServerConfig `yaml:"server"` + Database DatabaseConfig `yaml:"db"` + Elastic ElasticConfig `yaml:"elastic"` + VictoriaMetrics VictoriaMetricsConfig `yaml:"victoriametrics"` + OauthSigners []string `yaml:"openid"` + Clusters []ClusterConfig `yaml:"clusters"` + Security SecurityConfig `yaml:"security"` + RedisConfig RedisConfig `yaml:"redis"` } func ReadConfig(path string) *Config { @@ -70,6 +75,9 @@ func ReadConfig(path string) *Config { if envvar, ok := os.LookupEnv("REDIS_PASSWORD"); ok { config.RedisConfig.Password = envvar } + if envvar, ok := os.LookupEnv("VICTORIAMETRICS_TOKEN"); ok { + config.VictoriaMetrics.Token = envvar + } if config.Database.Address == "" || config.Database.User == "" || @@ -84,6 +92,11 @@ func ReadConfig(path string) *Config { log.Fatalf("Elastic config section does not pass sanity checks. URL, Username and Password must all not be empty") } + if config.VictoriaMetrics.URL == "" || + config.VictoriaMetrics.Token == "" { + log.Fatalf("VictoriaMetrics config section does not pass sanity checks. URL and Token must both not be empty") + } + if len(config.OauthSigners) == 0 { log.Fatalf("OpenId config section does not pass sanity checks. It must contain at least one certificates URL which signs JWTs") } diff --git a/victoriametrics/victoriametrics.go b/victoriametrics/victoriametrics.go new file mode 100644 index 0000000..6d70188 --- /dev/null +++ b/victoriametrics/victoriametrics.go @@ -0,0 +1,197 @@ +package victoriametrics + +import ( + "encoding/json" + "fmt" + "math" + "net/url" + "regexp" + "strconv" + "time" + + "github.com/rs/zerolog" + + "cscs.ch/hpcdata/elastic" + "cscs.ch/hpcdata/logging" + "cscs.ch/hpcdata/util" +) + +const wanted_num_timestamps = 5000 + +type MetricInfo struct { + VMMetric string // full Prometheus metric name as stored in VictoriaMetrics + Unit string // metric unit +} + +var SupportedMetrics = map[string]MetricInfo{ + "gpu_temp": {VMMetric: "cray_storage.dcgm.gpu_temp", Unit: "°C"}, + "memory_temp": {VMMetric: "cray_storage.dcgm.memory_temp", Unit: "°C"}, + "power_usage": {VMMetric: "cray_storage.dcgm.power_usage", Unit: "W"}, + "gpu_utilization": {VMMetric: "cray_storage.dcgm.gpu_utilization", Unit: "%"}, +} + +type Client struct { + baseURL string + token string +} + +type queryResponse struct { + Status string `json:"status"` + Data struct { + ResultType string `json:"resultType"` + Result []querySeries `json:"result"` + } `json:"data"` +} + +type querySeries struct { + Metric map[string]string `json:"metric"` + Values [][]interface{} `json:"values"` +} + +func NewClient(config *util.Config) *Client { + return &Client{ + baseURL: config.VictoriaMetrics.URL, + token: config.VictoriaMetrics.Token, + } +} + +func (c *Client) GetDcgmMetric(metricName string, nodes []util.Node, from time.Time, to time.Time, vcluster string, logger *zerolog.Logger) (*elastic.DcgmMetric, error) { + if logger == nil { + logger = logging.Get() + } + + info, ok := SupportedMetrics[metricName] + if !ok { + return nil, fmt.Errorf("Unsupported VM DCGM metric: %v", metricName) + } + + step := getStep(from, to, 10*time.Second) // [TO DO] avoid hard-coding the step duration + query := buildNodeRegexQuery(info.VMMetric, vcluster, nodes) + + logger.Debug().Msgf("VictoriaMetrics metric=%v query=%v start=%v end=%v step=%v", metricName, query, from.Unix(), to.Unix(), step) + + resp, err := c.queryRange(query, from, to, step, logger) + if err != nil { + return nil, fmt.Errorf("Failed querying VictoriaMetrics for metric %v: %w", metricName, err) + } + + if resp.Status != "success" { + return nil, fmt.Errorf("VictoriaMetrics query for metric %v returned non-success status: %v", metricName, resp.Status) + } + + stepDuration, err := time.ParseDuration(step) + if err != nil { + return nil, fmt.Errorf("Failed parsing step duration %v: %w", step, err) + } + + ret := elastic.DcgmMetric{MetricByNode: map[string][]elastic.DcgmDataIndexed{}} + + if len(resp.Data.Result) == 0 { + return &ret, nil + } + + // Build a time grid aligned to the requested step + numPoints := int(to.Sub(from).Seconds()/stepDuration.Seconds()) + 1 + ret.Time = make([]time.Time, numPoints) + for i := 0; i < numPoints; i++ { + ret.Time[i] = from.Add(time.Duration(i) * stepDuration) + } + + for _, series := range resp.Data.Result { + hostname := series.Metric["hostname"] + gpuIDStr := series.Metric["gpu_id"] + gpuID, err := strconv.Atoi(gpuIDStr) + if err != nil { + return nil, fmt.Errorf("Failed parsing gpu_id=%v as integer", gpuIDStr) + } + + // Ensure the per-GPU slot exists and is zero-filled for the whole grid + for len(ret.MetricByNode[hostname]) <= gpuID { + ret.MetricByNode[hostname] = append(ret.MetricByNode[hostname], elastic.DcgmDataIndexed{}) + ret.MetricByNode[hostname][len(ret.MetricByNode[hostname])-1].Data = make([]float64, numPoints) + } + ret.MetricByNode[hostname][gpuID].GpuIndex = gpuID + + for _, pair := range series.Values { + if len(pair) != 2 { + continue + } + tsFloat, ok := pair[0].(float64) + if !ok { + continue + } + valStr, ok := pair[1].(string) + if !ok { + continue + } + val, err := strconv.ParseFloat(valStr, 64) + if err != nil { + logger.Warn().Msgf("Failed parsing value %v as float64, skipping sample", valStr) + continue + } + ts := time.Unix(int64(tsFloat), 0) + idx := int(ts.Sub(from).Seconds() / stepDuration.Seconds()) + if idx >= 0 && idx < numPoints { + ret.MetricByNode[hostname][gpuID].Data[idx] = val + } + } + } + + return &ret, nil +} + +func (c *Client) queryRange(query string, from time.Time, to time.Time, step string, logger *zerolog.Logger) (*queryResponse, error) { + params := url.Values{} + params.Set("query", query) + params.Set("start", strconv.FormatInt(from.Unix(), 10)) + params.Set("end", strconv.FormatInt(to.Unix(), 10)) + params.Set("step", step) + + requestURL := fmt.Sprintf("%v/select/0/prometheus/api/v1/query_range?%v", c.baseURL, params.Encode()) + headers := map[string]string{ + "Authorization": fmt.Sprintf("Bearer %v", c.token), + } + + resp, err := util.DoRequest("GET", requestURL, headers, nil) + if err != nil { + return nil, err + } + if err := util.CheckResponse(resp); err != nil { + return nil, err + } + + logger.Debug().Msgf("VictoriaMetrics query_range response: %v", string(resp.ResponseData)) + + var parsed queryResponse + if err := json.Unmarshal(resp.ResponseData, &parsed); err != nil { + return nil, fmt.Errorf("Failed unmarshaling VictoriaMetrics response: %w", err) + } + return &parsed, nil +} + +// build a PromQL query that filters by vcluster and by a regex list of hostnames +func buildNodeRegexQuery(metric string, vcluster string, nodes []util.Node) string { + hostnames := make([]string, 0, len(nodes)) + for _, n := range nodes { + hostnames = append(hostnames, regexp.QuoteMeta(n.Nid)) + } + regex := "" + for i, h := range hostnames { + if i > 0 { + regex += "|" + } + regex += h + } + return fmt.Sprintf("%v{vcluster=\"%v\", hostname=~\"%v\"}", metric, vcluster, regex) +} + +// This is a rewrite of elastic.go get_interval +// [TO DO]: avoid duplication +func getStep(from, to time.Time, minStep time.Duration) string { + totalSec := to.Sub(from).Seconds() + autoStepSec := math.Ceil(totalSec / wanted_num_timestamps) + if autoStepSec < minStep.Seconds() { + return fmt.Sprintf("%vs", math.Ceil(minStep.Seconds())) + } + return fmt.Sprintf("%vs", autoStepSec) +} From 73b2d07995bb578469eb4484266b81dd768a1c2c Mon Sep 17 00:00:00 2001 From: sgozel Date: Wed, 14 Oct 2026 10:48:35 +0200 Subject: [PATCH 2/6] stop logging the entire VM query response --- victoriametrics/victoriametrics.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/victoriametrics/victoriametrics.go b/victoriametrics/victoriametrics.go index 6d70188..8c16f9a 100644 --- a/victoriametrics/victoriametrics.go +++ b/victoriametrics/victoriametrics.go @@ -160,8 +160,6 @@ func (c *Client) queryRange(query string, from time.Time, to time.Time, step str return nil, err } - logger.Debug().Msgf("VictoriaMetrics query_range response: %v", string(resp.ResponseData)) - var parsed queryResponse if err := json.Unmarshal(resp.ResponseData, &parsed); err != nil { return nil, fmt.Errorf("Failed unmarshaling VictoriaMetrics response: %w", err) From 7a8c52d771b219b2289e0293ba8a218e03921ed5 Mon Sep 17 00:00:00 2001 From: sgozel Date: Wed, 14 Oct 2026 11:55:37 +0200 Subject: [PATCH 3/6] add check of VM query result type (should be matrix) --- victoriametrics/victoriametrics.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/victoriametrics/victoriametrics.go b/victoriametrics/victoriametrics.go index 8c16f9a..bf4a627 100644 --- a/victoriametrics/victoriametrics.go +++ b/victoriametrics/victoriametrics.go @@ -164,6 +164,11 @@ func (c *Client) queryRange(query string, from time.Time, to time.Time, step str if err := json.Unmarshal(resp.ResponseData, &parsed); err != nil { return nil, fmt.Errorf("Failed unmarshaling VictoriaMetrics response: %w", err) } + + if parsed.Data.ResultType != "matrix" { + return nil, fmt.Errorf("Unexpected result type different from matrix: %q", parsed.Data.ResultType) + } + return &parsed, nil } From 85dce781c20677d160baeb58a038dee3e297181c Mon Sep 17 00:00:00 2001 From: sgozel Date: Wed, 14 Oct 2026 12:09:33 +0200 Subject: [PATCH 4/6] delegate VM response status check to queryRange rather than GetDcgmMetric --- victoriametrics/victoriametrics.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/victoriametrics/victoriametrics.go b/victoriametrics/victoriametrics.go index bf4a627..5cced3c 100644 --- a/victoriametrics/victoriametrics.go +++ b/victoriametrics/victoriametrics.go @@ -75,10 +75,6 @@ func (c *Client) GetDcgmMetric(metricName string, nodes []util.Node, from time.T return nil, fmt.Errorf("Failed querying VictoriaMetrics for metric %v: %w", metricName, err) } - if resp.Status != "success" { - return nil, fmt.Errorf("VictoriaMetrics query for metric %v returned non-success status: %v", metricName, resp.Status) - } - stepDuration, err := time.ParseDuration(step) if err != nil { return nil, fmt.Errorf("Failed parsing step duration %v: %w", step, err) @@ -165,6 +161,10 @@ func (c *Client) queryRange(query string, from time.Time, to time.Time, step str return nil, fmt.Errorf("Failed unmarshaling VictoriaMetrics response: %w", err) } + if parsed.Status != "success" { + return nil, fmt.Errorf("VictoriaMetrics query_range returned non-success status %q for query %q", parsed.Status, query) + } + if parsed.Data.ResultType != "matrix" { return nil, fmt.Errorf("Unexpected result type different from matrix: %q", parsed.Data.ResultType) } From 8ea355633c2e47ad3960af7ea60fb631fad782f1 Mon Sep 17 00:00:00 2001 From: sgozel Date: Mon, 19 Oct 2026 13:51:23 +0200 Subject: [PATCH 5/6] resolve conversation --- handler/vm_dcgm.go | 4 ++-- victoriametrics/victoriametrics.go | 17 ++++++++--------- 2 files changed, 10 insertions(+), 11 deletions(-) diff --git a/handler/vm_dcgm.go b/handler/vm_dcgm.go index 3a0c358..f521d78 100644 --- a/handler/vm_dcgm.go +++ b/handler/vm_dcgm.go @@ -34,7 +34,7 @@ func (h vmDcgm) Get(w http.ResponseWriter, r *http.Request) { info, ok := victoriametrics.SupportedMetrics[metricName] if !ok { - pie(logger.Warn, condition_error{fmt.Sprintf("Unsupported VM DCGM metric: %v", metricName)}, "", http.StatusBadRequest) + pie(logger.Warn, condition_error{fmt.Sprintf("Unsupported DCGM metric: %v", metricName)}, "", http.StatusBadRequest) } logger.Debug().Msgf("Passed all security checks to fetch VM DCGM metric=%v for job=%+v in the time window from=%v to=%v", metricName, job, from, to) @@ -53,7 +53,7 @@ func (h vmDcgm) Get(w http.ResponseWriter, r *http.Request) { } dcgmData, err := h.vmclient.GetDcgmMetric(metricName, nodes, from, to, cluster_config.Name, logger) - pie(logger.Error, err, fmt.Sprintf("Failed getting VM DCGM metric %v", metricName), http.StatusInternalServerError) + pie(logger.Error, err, fmt.Sprintf("Failed getting DCGM metric %v", metricName), http.StatusInternalServerError) ret := struct { Time []epochTime `json:"time"` diff --git a/victoriametrics/victoriametrics.go b/victoriametrics/victoriametrics.go index 5cced3c..5390433 100644 --- a/victoriametrics/victoriametrics.go +++ b/victoriametrics/victoriametrics.go @@ -7,6 +7,7 @@ import ( "net/url" "regexp" "strconv" + "strings" "time" "github.com/rs/zerolog" @@ -62,7 +63,7 @@ func (c *Client) GetDcgmMetric(metricName string, nodes []util.Node, from time.T info, ok := SupportedMetrics[metricName] if !ok { - return nil, fmt.Errorf("Unsupported VM DCGM metric: %v", metricName) + return nil, fmt.Errorf("Unsupported DCGM metric: %v", metricName) } step := getStep(from, to, 10*time.Second) // [TO DO] avoid hard-coding the step duration @@ -114,10 +115,12 @@ func (c *Client) GetDcgmMetric(metricName string, nodes []util.Node, from time.T } tsFloat, ok := pair[0].(float64) if !ok { + logger.Warn().Msgf("Failed parsing timestamp %v as float64, skipping sample", pair[0]) continue } valStr, ok := pair[1].(string) if !ok { + logger.Warn().Msgf("Failed parsing value %v as string, skipping sample", pair[1]) continue } val, err := strconv.ParseFloat(valStr, 64) @@ -129,6 +132,8 @@ func (c *Client) GetDcgmMetric(metricName string, nodes []util.Node, from time.T idx := int(ts.Sub(from).Seconds() / stepDuration.Seconds()) if idx >= 0 && idx < numPoints { ret.MetricByNode[hostname][gpuID].Data[idx] = val + } else { + logger.Warn().Msgf("Time index %v falling out of bounds [%v, %v[", idx, 0, numPoints) } } } @@ -178,14 +183,8 @@ func buildNodeRegexQuery(metric string, vcluster string, nodes []util.Node) stri for _, n := range nodes { hostnames = append(hostnames, regexp.QuoteMeta(n.Nid)) } - regex := "" - for i, h := range hostnames { - if i > 0 { - regex += "|" - } - regex += h - } - return fmt.Sprintf("%v{vcluster=\"%v\", hostname=~\"%v\"}", metric, vcluster, regex) + regex := strings.Join(hostnames, "|") + return fmt.Sprintf(`%v{vcluster="%v", hostname=~"%v"}`, metric, vcluster, regex) } // This is a rewrite of elastic.go get_interval From d4250b67554b1fee180c8651e31dea57a4cc07a2 Mon Sep 17 00:00:00 2001 From: sgozel Date: Fri, 23 Oct 2026 14:19:13 +0200 Subject: [PATCH 6/6] log VM request URL logger input argument was unused. Now logs the victoriametrics request URL. --- victoriametrics/victoriametrics.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/victoriametrics/victoriametrics.go b/victoriametrics/victoriametrics.go index 5390433..2bdea19 100644 --- a/victoriametrics/victoriametrics.go +++ b/victoriametrics/victoriametrics.go @@ -142,6 +142,10 @@ func (c *Client) GetDcgmMetric(metricName string, nodes []util.Node, from time.T } func (c *Client) queryRange(query string, from time.Time, to time.Time, step string, logger *zerolog.Logger) (*queryResponse, error) { + if logger == nil { + logger = logging.Get() + } + params := url.Values{} params.Set("query", query) params.Set("start", strconv.FormatInt(from.Unix(), 10)) @@ -149,6 +153,8 @@ func (c *Client) queryRange(query string, from time.Time, to time.Time, step str params.Set("step", step) requestURL := fmt.Sprintf("%v/select/0/prometheus/api/v1/query_range?%v", c.baseURL, params.Encode()) + logger.Debug().Msgf("VictoriaMetrics query_range URL=%v", requestURL) + headers := map[string]string{ "Authorization": fmt.Sprintf("Bearer %v", c.token), }