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..f521d78 --- /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 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 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..2bdea19 --- /dev/null +++ b/victoriametrics/victoriametrics.go @@ -0,0 +1,205 @@ +package victoriametrics + +import ( + "encoding/json" + "fmt" + "math" + "net/url" + "regexp" + "strconv" + "strings" + "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 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) + } + + 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 { + 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) + 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 + } else { + logger.Warn().Msgf("Time index %v falling out of bounds [%v, %v[", idx, 0, numPoints) + } + } + } + + return &ret, nil +} + +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)) + 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()) + logger.Debug().Msgf("VictoriaMetrics query_range URL=%v", requestURL) + + 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 + } + + var parsed queryResponse + if err := json.Unmarshal(resp.ResponseData, &parsed); err != nil { + 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) + } + + 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 := strings.Join(hostnames, "|") + 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) +}