Merge pull request #60 from castai/feat-additional-gpu-metrics

feat: Start collecting more GPU metrics

Author: Ivaka

.mockery.yaml

@@ -1,12 +1,18 @@
-with-expecter: true
-dir: 'mock/{{replace .InterfaceDirRelative "internal" "" 1}}'
+dir: mock/{{.InterfaceDirRelative | replace "internal" "" 1}}
+filename: "mock_{{.InterfaceName | lower}}.go"
+template: testify
+template-data:
+  unroll-variadic: true
 packages:
   "github.com/castai/gpu-metrics-exporter/internal/exporter":
     interfaces:
-      Exporter:
-      Scraper:
-      MetricMapper:
-      HttpClient:
+      Exporter: {}
+      Scraper: {}
+      MetricMapper: {}
+      HTTPClient: {}
+  "github.com/castai/gpu-metrics-exporter/internal/workload":
+    interfaces:
+      Resolver: {}
   "github.com/castai/gpu-metrics-exporter/internal/castai":
     interfaces:
-      Client:
+      Client: {}

README.md

@@ -20,21 +20,38 @@ nv-hostengine.
 Make sure that these fields are exposed by DCGM exporter as metrics:
 
 ```
+DCGM_FI_DEV_GPU_TEMP
+DCGM_FI_DEV_MEMORY_TEMP
+DCGM_FI_DEV_POWER_USAGE
+DCGM_FI_DEV_MEM_MAX_OP_TEMP
+DCGM_FI_DEV_GPU_MAX_OP_TEMP
+DCGM_FI_DEV_SM_CLOCK
+DCGM_FI_DEV_GPU_UTIL
 DCGM_FI_PROF_SM_ACTIVE
 DCGM_FI_PROF_SM_OCCUPANCY
-DCGM_FI_PROF_PIPE_TENSOR_ACTIVE
-DCGM_FI_PROF_DRAM_ACTIVE
-DCGM_FI_PROF_PCIE_TX_BYTES
-DCGM_FI_PROF_PCIE_RX_BYTES
 DCGM_FI_PROF_GR_ENGINE_ACTIVE
-DCGM_FI_DEV_FB_TOTAL
+DCGM_FI_PROF_DRAM_ACTIVE
 DCGM_FI_DEV_FB_FREE
 DCGM_FI_DEV_FB_USED
+DCGM_FI_DEV_FB_TOTAL
+DCGM_FI_DEV_MEM_COPY_UTIL
+DCGM_FI_PROF_PCIE_TX_BYTES
+DCGM_FI_PROF_PCIE_RX_BYTES
 DCGM_FI_DEV_PCIE_LINK_GEN
 DCGM_FI_DEV_PCIE_LINK_WIDTH
-DCGM_FI_DEV_GPU_TEMP
-DCGM_FI_DEV_MEMORY_TEMP
-DCGM_FI_DEV_POWER_USAGE
+DCGM_FI_PROF_NVLINK_TX_BYTES
+DCGM_FI_PROF_NVLINK_RX_BYTES
+DCGM_FI_PROF_PIPE_INT_ACTIVE
+DCGM_FI_PROF_PIPE_FP16_ACTIVE
+DCGM_FI_PROF_PIPE_FP32_ACTIVE
+DCGM_FI_PROF_PIPE_FP64_ACTIVE
+DCGM_FI_PROF_PIPE_TENSOR_ACTIVE
+DCGM_FI_DEV_MIG_MODE
+DCGM_FI_DEV_MIG_MAX_SLICES
+DCGM_FI_DEV_CLOCKS_EVENT_REASONS
+DCGM_FI_DEV_XID_ERRORS
+DCGM_FI_DEV_POWER_VIOLATION
+DCGM_FI_DEV_THERMAL_VIOLATION
 ```
 
 ## Installation

charts/gpu-metrics-exporter/templates/dcgm-exporter-configmap.yaml

@@ -16,14 +16,14 @@ data:
     DCGM_FI_DEV_GPU_UTIL, gauge, GPU utilization (in %).
     DCGM_FI_PROF_SM_ACTIVE, gauge, The ratio of cycles an SM has at least 1 warp assigned
     DCGM_FI_PROF_SM_OCCUPANCY, gauge, The fraction of resident warps on a multiprocessor
-    DCGM_FI_PROF_PIPE_TENSOR_ACTIVE, gauge, The ratio of cycles the tensor (HMMA) pipe is active (off the peak sustained elapsed cycles)
     DCGM_FI_PROF_GR_ENGINE_ACTIVE, gauge, Ratio of time the graphics engine is active (in %).
     DCGM_FI_PROF_DRAM_ACTIVE, gauge, The ratio of cycles the device memory interface is active sending or receiving data.
 
     # Memory usage,,
     DCGM_FI_DEV_FB_FREE, gauge, Framebuffer memory free (in MiB).
     DCGM_FI_DEV_FB_USED, gauge, Framebuffer memory used (in MiB).
     DCGM_FI_DEV_FB_TOTAL, gauge, Total Frame Buffer of the GPU in MB.
+    DCGM_FI_DEV_MEM_COPY_UTIL, gauge, Utilization of the memory copy engine.
 
     # PCIE,,
     DCGM_FI_PROF_PCIE_TX_BYTES, gauge, Total number of bytes transmitted through PCIe TX
@@ -36,4 +36,15 @@ data:
     DCGM_FI_PROF_PIPE_FP16_ACTIVE, gauge, Ratio of cycles the fp16 pipe is active.
     DCGM_FI_PROF_PIPE_FP32_ACTIVE, gauge, Ratio of cycles the fp32 pipe is active.
     DCGM_FI_PROF_PIPE_FP64_ACTIVE, gauge, Ratio of cycles the fp64 pipe is active.
+    DCGM_FI_PROF_PIPE_TENSOR_ACTIVE, gauge, The ratio of cycles the tensor (HMMA) pipe is active (off the peak sustained elapsed cycles)
+
+    # Health,,
+    DCGM_FI_DEV_CLOCKS_EVENT_REASONS, gauge, Current clock event reasons (bitmask of DCGM_CLOCKS_EVENT_REASON_*)
+    DCGM_FI_DEV_XID_ERRORS, gauge, The value is the specific XID error
+    DCGM_FI_DEV_POWER_VIOLATION, gauge, Power Violation time in ns.
+    DCGM_FI_DEV_THERMAL_VIOLATION, gauge, Thermal Violation time in ns.
+
+    # NVLink,,
+    DCGM_FI_PROF_NVLINK_TX_BYTES, gauge, The number of bytes of active NvLink tx (transmit) data including both header and payload.
+    DCGM_FI_PROF_NVLINK_RX_BYTES, gauge, The number of bytes of active NvLink rx (read) data including both header and payload.
 {{- end }}

charts/gpu-metrics-exporter/templates/move-api-key-to-secret.yaml

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"net/http"
 	"os"
 	"os/signal"
@@ -14,14 +15,17 @@ import (
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/selection"
-	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/util/flowcontrol"
 
 	"github.com/castai/gpu-metrics-exporter/internal/castai"
 	"github.com/castai/gpu-metrics-exporter/internal/config"
 	"github.com/castai/gpu-metrics-exporter/internal/exporter"
 	"github.com/castai/gpu-metrics-exporter/internal/server"
+	"github.com/castai/gpu-metrics-exporter/internal/workload"
+	"github.com/castai/logging"
+	"github.com/castai/metrics"
 )
 
 var (
@@ -30,6 +34,11 @@ var (
 	Version   = "local"
 )
 
+const (
+	workloadCacheSize = 512
+	workloadsLabelKey = "workloads.cast.ai/custom-workload"
+)
+
 func main() {
 	log := logrus.New()
 
@@ -38,19 +47,30 @@ func main() {
 		log.Fatal(err)
 	}
 
-	logLevel, err := logrus.ParseLevel(cfg.LogLevel)
+	logLevel, err := parseLogLevel(cfg.LogLevel)
 	if err != nil {
-		log.Fatal(err)
+		log.Warnf("failed to parse log level, defaulting to 'info': %v", err)
+		logLevel = slog.LevelInfo
 	}
-	log.SetLevel(logLevel)
 
-	if err := run(cfg, log); err != nil && !errors.Is(err, context.Canceled) {
+	castaiLogger := logging.New(logging.NewTextHandler(logging.TextHandlerConfig{
+		Output: os.Stdout,
+		Level:  logLevel,
+	}))
+
+	if err := run(cfg, castaiLogger); err != nil && !errors.Is(err, context.Canceled) {
 		log.Fatal(err)
 	}
 }
 
-func run(cfg *config.Config, log logrus.FieldLogger) error {
-	mux := server.NewServerMux(log)
+func parseLogLevel(level string) (slog.Level, error) {
+	var lvl slog.Level
+	err := lvl.UnmarshalText([]byte(level))
+	return lvl, err
+}
+
+func run(cfg *config.Config, log *logging.Logger) error {
+	mux := server.NewServerMux()
 
 	srv := &http.Server{
 		Addr:              fmt.Sprintf(":%d", cfg.HTTPListenPort),
@@ -75,19 +95,45 @@ func run(cfg *config.Config, log logrus.FieldLogger) error {
 		cancel()
 	}()
 
-	clientset, err := newKubernetesClientset(cfg)
+	dynClient, err := newDynamicClient(cfg)
 	if err != nil {
-		log.Fatal(err)
+		log.WithField("error", err.Error()).Fatal("failed to create kubernetes dynamic client")
 	}
 
 	labelSelector, err := selectorFromMap(cfg.DCGMLabels)
 	if err != nil {
-		log.Fatal(err)
+		log.WithField("error", err.Error()).Fatal("failed to create get label selector")
+	}
+
+	metricClient, err := metrics.NewMetricClient(
+		metrics.Config{
+			APIAddr:   cfg.TelemetryURL,
+			APIToken:  cfg.APIKey,
+			ClusterID: cfg.ClusterID,
+		}, log)
+	if err != nil {
+		log.WithField("error", err.Error()).Warn("failed to create metrics client")
+	}
+
+	if metricClient != nil {
+		go func() {
+			if err := metricClient.Start(ctx); err != nil && !errors.Is(err, context.Canceled) {
+				log.WithField("error", err.Error()).Error("error in metrics client")
+			}
+		}()
 	}
 
 	client := setupCastAIClient(log, cfg)
 	scraper := exporter.NewScraper(&http.Client{}, log)
-	mapper := exporter.NewMapper(cfg.NodeName)
+	workloadResolver, err := workload.NewResolver(dynClient, workload.Config{
+		LabelKeys: []string{workloadsLabelKey},
+		CacheSize: workloadCacheSize,
+	})
+	if err != nil {
+		log.WithField("error", err.Error()).Fatal("failed to create workload resolver")
+	}
+
+	mapper := exporter.NewMapper(cfg.NodeName, workloadResolver, log)
 	ex := exporter.NewExporter(exporter.Config{
 		ExportInterval:   cfg.ExportInterval,
 		Selector:         labelSelector.String(),
@@ -96,7 +142,7 @@ func run(cfg *config.Config, log logrus.FieldLogger) error {
 		DCGMExporterHost: cfg.DCGMHost,
 		Enabled:          true,
 		NodeName:         cfg.NodeName,
-	}, clientset, log, scraper, mapper, client)
+	}, dynClient, log, scraper, mapper, client, metricClient)
 
 	go func() {
 		if err := ex.Start(ctx); err != nil && !errors.Is(err, context.Canceled) {
@@ -108,19 +154,14 @@ func run(cfg *config.Config, log logrus.FieldLogger) error {
 	return srv.ListenAndServe()
 }
 
-func newKubernetesClientset(cfg *config.Config) (*kubernetes.Clientset, error) {
-	config, err := clientcmd.BuildConfigFromFlags("", cfg.KubeConfigPath)
-	if err != nil {
-		return nil, err
-	}
-	config.RateLimiter = flowcontrol.NewTokenBucketRateLimiter(float32(10), 25)
-
-	clientset, err := kubernetes.NewForConfig(config)
+func newDynamicClient(cfg *config.Config) (dynamic.Interface, error) {
+	restConfig, err := clientcmd.BuildConfigFromFlags("", cfg.KubeConfigPath)
 	if err != nil {
 		return nil, err
 	}
+	restConfig.RateLimiter = flowcontrol.NewTokenBucketRateLimiter(float32(10), 25)
 
-	return clientset, nil
+	return dynamic.NewForConfig(restConfig)
 }
 
 func selectorFromMap(labelMap map[string]string) (labels.Selector, error) {
@@ -138,7 +179,7 @@ func selectorFromMap(labelMap map[string]string) (labels.Selector, error) {
 	return selector.Add(requirements...), nil
 }
 
-func setupCastAIClient(log logrus.FieldLogger, cfg *config.Config) castai.Client {
+func setupCastAIClient(log *logging.Logger, cfg *config.Config) castai.Client {
 	clientConfig := castai.Config{
 		ClusterID: cfg.ClusterID,
 		APIKey:    cfg.APIKey,

cmd/main.go

@@ -1,2 +1,2 @@
-//go:generate go run github.com/vektra/mockery/v2@v2.42.0 --all
+//go:generate go run github.com/vektra/mockery/v3@v3.5.3
 package mockery