feat(monitor): Standardized OpenTelemetry and Dynamic Cluster Node Labeling

CHANGELOG.md

@@ -403,6 +403,7 @@ Add "NVIDIA_VISIBLE_DEVICES=none" to none-gpu tasks
 - Add Ascend core resource for HAMi-vNPU-core virtualization
 - Add enableGetPreferredAllocation flag
 - Add local-deploy target for minikube/kind clusters
+- Standardize OpenTelemetry metrics and dynamic cluster node labeling in vGPUmonitor
 
 **Bug fixes**
 - Fix initialization error when using tensor parallelism on vLLM above 0.18

cmd/vGPUmonitor/main.go

@@ -52,6 +52,8 @@ var (
 	}
 	metricsBindAddress string
 	legacyMetrics      bool
+	nodeName           string
+	clusterID          string
 )
 
 func init() {
@@ -60,6 +62,8 @@ func init() {
 	rootCmd.Flags().AddGoFlagSet(util.InitKlogFlags())
 	rootCmd.Flags().StringVar(&metricsBindAddress, "metrics-bind-address", ":9394", "The TCP address that the vGPUmonitor should bind to for serving prometheus metrics(e.g. 127.0.0.1:9394, :9394)")
 	rootCmd.Flags().BoolVar(&legacyMetrics, "legacy-metrics", false, "Emit legacy metric names alongside new ones for backward compatibility")
+	rootCmd.Flags().StringVar(&nodeName, "node-name", "", "The name of the node where vGPUmonitor is running")
+	rootCmd.Flags().StringVar(&clusterID, "cluster-id", "", "The ID of the cluster")
 	rootCmd.AddCommand(version.VersionCmd)
 }
 
@@ -142,7 +146,7 @@ func initMetrics(ctx context.Context, containerLister *nvidia.ContainerLister) e
 
 	reg.MustRegister(versionmetrics.NewBuildInfoCollector())
 
-	NewClusterManager("vGPU", reg, containerLister, legacyMetrics)
+	NewClusterManager("vGPU", reg, containerLister, legacyMetrics, nodeName, clusterID)
 
 	// Uncomment to add the standard process and Go metrics to the custom registry.
 	//reg.MustRegister(

cmd/vGPUmonitor/metrics.go

@@ -47,6 +47,8 @@ type ClusterManager struct {
 	PodLister       listerscorev1.PodLister
 	containerLister *nvidia.ContainerLister
 	LegacyMetrics   bool
+	NodeName        string
+	ClusterID       string
 }
 
 // ClusterManagerCollector implements the Collector interface.
@@ -360,7 +362,10 @@ func (cc ClusterManagerCollector) collectGPUUtilizationMetrics(ch chan<- prometh
 }
 
 func (cc ClusterManagerCollector) collectPodAndContainerInfo(ch chan<- prometheus.Metric) error {
-	nodeName := os.Getenv(util.NodeNameEnvName)
+	nodeName := cc.ClusterManager.NodeName
+	if nodeName == "" {
+		nodeName = os.Getenv(util.NodeNameEnvName)
+	}
 	if nodeName == "" {
 		return fmt.Errorf("node name environment variable %s is not set", util.NodeNameEnvName)
 	}
@@ -504,7 +509,10 @@ func (cc ClusterManagerCollector) collectContainerMetrics(ch chan<- prometheus.M
 }
 
 func (cc ClusterManagerCollector) collectPodAndContainerMigInfo(ch chan<- prometheus.Metric) error {
-	nodeName := os.Getenv(util.NodeNameEnvName)
+	nodeName := cc.ClusterManager.NodeName
+	if nodeName == "" {
+		nodeName = os.Getenv(util.NodeNameEnvName)
+	}
 	if nodeName == "" {
 		return fmt.Errorf("node name environment variable %s is not set", util.NodeNameEnvName)
 	}
@@ -563,23 +571,39 @@ func sendMetric(ch chan<- prometheus.Metric, desc *prometheus.Desc, valueType pr
 
 // NewClusterManager creates a ClusterManager for the given zone, backs its pod
 // lookups with a shared informer, and registers its collector with reg through
-// a wrapping Registerer that adds the zone as a label.
-func NewClusterManager(zone string, reg prometheus.Registerer, containerLister *nvidia.ContainerLister, legacyMetrics bool) *ClusterManager {
+// a wrapping Registerer that adds the zone, node_name, and cluster_id as labels.
+func NewClusterManager(zone string, reg prometheus.Registerer, containerLister *nvidia.ContainerLister, legacyMetrics bool, nodeName string, clusterID string) *ClusterManager {
 	if legacyMetrics {
 		initLegacyDescriptors()
 	}
+	if nodeName == "" {
+		nodeName = os.Getenv(util.NodeNameEnvName)
+	}
 	c := &ClusterManager{
 		Zone:            zone,
 		containerLister: containerLister,
 		LegacyMetrics:   legacyMetrics,
+		NodeName:        nodeName,
+		ClusterID:       clusterID,
 	}
 
-	informerFactory := informers.NewSharedInformerFactoryWithOptions(containerLister.Clientset(), time.Hour*1)
-	c.PodLister = informerFactory.Core().V1().Pods().Lister()
-	stopCh := make(chan struct{})
-	informerFactory.Start(stopCh)
+	if containerLister != nil && containerLister.Clientset() != nil {
+		informerFactory := informers.NewSharedInformerFactoryWithOptions(containerLister.Clientset(), time.Hour*1)
+		c.PodLister = informerFactory.Core().V1().Pods().Lister()
+		stopCh := make(chan struct{})
+		informerFactory.Start(stopCh)
+	}
 
 	cc := ClusterManagerCollector{ClusterManager: c}
-	prometheus.WrapRegistererWith(prometheus.Labels{"zone": zone}, reg).MustRegister(cc)
+	labels := prometheus.Labels{
+		"zone": zone,
+	}
+	if nodeName != "" {
+		labels["node_name"] = nodeName
+	}
+	if clusterID != "" {
+		labels["cluster_id"] = clusterID
+	}
+	prometheus.WrapRegistererWith(labels, reg).MustRegister(cc)
 	return c
 }

cmd/vGPUmonitor/metrics_test.go

@@ -68,3 +68,16 @@ func TestDescribeCollectSync(t *testing.T) {
 		t.Errorf("Gather failed (legacy): %v", err)
 	}
 }
+
+func TestNewClusterManagerNodeAndClusterIDLabels(t *testing.T) {
+	reg := prometheus.NewRegistry()
+	client := fake.NewSimpleClientset()
+	containerLister := nvidia.NewContainerListerWithClientset(client)
+	c := NewClusterManager("vGPU", reg, containerLister, false, "node-1", "cluster-1")
+	if c.NodeName != "node-1" {
+		t.Errorf("expected NodeName node-1, got %s", c.NodeName)
+	}
+	if c.ClusterID != "cluster-1" {
+		t.Errorf("expected ClusterID cluster-1, got %s", c.ClusterID)
+	}
+}

pkg/monitor/nvidia/cudevshr.go

@@ -84,7 +84,7 @@ type ContainerLister struct {
 	containerPath string
 	containers    map[string]*ContainerUsage
 	mutex         sync.Mutex
-	clientset     *kubernetes.Clientset
+	clientset     kubernetes.Interface
 	nodeName      string
 
 	// Fields for the informer-based pod cache mechanism
@@ -157,7 +157,15 @@ func (l *ContainerLister) ListContainers() map[string]*ContainerUsage {
 	return l.containers
 }
 
-func (l *ContainerLister) Clientset() *kubernetes.Clientset {
+// NewContainerListerWithClientset creates a ContainerLister initialized with the given clientset.
+func NewContainerListerWithClientset(clientset kubernetes.Interface) *ContainerLister {
+	return &ContainerLister{
+		containers: make(map[string]*ContainerUsage),
+		clientset:  clientset,
+	}
+}
+
+func (l *ContainerLister) Clientset() kubernetes.Interface {
 	return l.clientset
 }