Add VM endpoint and DCGM metrics

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

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

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:

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()

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()

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)
+}

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"))

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")
 	}