[EBPF] gpu: add NVLink fabric topology tags (#53721)

### What does this PR do?
Adds `gpu_fabric_cluster_uuid` and `gpu_fabric_clique_id` tags to GPU metrics when NVLink fabric registration is complete.

### Motivation
Allow grouping GPU workloads by NVLink P2P topology.

### Describe how you validated your changes

Unit tests added. Due to the API design of this function, exact mocking is not possible. We also tested on a NVlink-enabled node but it did not have a cluster UUID as it was just a single node.

### Additional Notes


Co-authored-by: guillermo.julian <guillermo.julian@datadoghq.com>

Author: gjulianm

comp/core/tagger/collectors/workloadmeta_extract.go

@@ -915,6 +915,10 @@ func ExtractGPUTags(gpu *workloadmeta.GPU, tagList *taglist.TagList) {
 	tagList.AddLow(tags.GPUArchitecture, strings.ToLower(gpu.Architecture))
 	tagList.AddLow(tags.GPUSlicingMode, gpu.SlicingMode())
 	tagList.AddLow(tags.GPUPCIBusID, strings.ToLower(gpu.PCIBusID))
+	if gpu.FabricClusterUUID != "" {
+		tagList.AddLow(tags.GPUFabricClusterUUID, strings.ToLower(gpu.FabricClusterUUID))
+		tagList.AddLow(tags.GPUFabricCliqueID, strconv.FormatUint(uint64(gpu.FabricCliqueID), 10))
+	}
 	if gpu.GPUType != "" {
 		tagList.AddLow(tags.GPUType, strings.ToLower(gpu.GPUType))
 	}

comp/core/tagger/collectors/workloadmeta_test.go

@@ -3847,10 +3847,12 @@ func TestHandleGPU(t *testing.T) {
 				EntityMeta: workloadmeta.EntityMeta{
 					Name: entityID.ID,
 				},
-				Vendor:   "nvidia",
-				Device:   "tesla-v100",
-				GPUType:  "v100",
-				PCIBusID: "0000:00:1e.0",
+				Vendor:            "nvidia",
+				Device:            "tesla-v100",
+				GPUType:           "v100",
+				PCIBusID:          "0000:00:1e.0",
+				FabricClusterUUID: "00112233-4455-6677-8899-aabbccddeeff",
+				FabricCliqueID:    7,
 			},
 			expected: []*types.TagInfo{
 				{
@@ -3866,6 +3868,8 @@ func TestHandleGPU(t *testing.T) {
 						"gpu_slicing_mode:none",
 						"gpu_parent_uuid:gpu-1234",
 						"gpu_pci_bus_id:0000:00:1e.0",
+						"gpu_fabric_cluster_uuid:00112233-4455-6677-8899-aabbccddeeff",
+						"gpu_fabric_clique_id:7",
 					},
 					StandardTags: []string{},
 				},

comp/core/tagger/tags/tags.go

@@ -155,6 +155,10 @@ const (
 	GPUParentGPUUUID = "gpu_parent_uuid"
 	// GPUPCIBusID is the tag for the PCI bus ID of the GPU
 	GPUPCIBusID = "gpu_pci_bus_id"
+	// GPUFabricClusterUUID is the tag for the NVLink fabric cluster UUID of the GPU
+	GPUFabricClusterUUID = "gpu_fabric_cluster_uuid"
+	// GPUFabricCliqueID is the tag for the NVLink fabric clique ID of the GPU
+	GPUFabricCliqueID = "gpu_fabric_clique_id"
 
 	// KubeArgoRollout is the tag for the Argo Rollout name
 	KubeArgoRollout = "kube_argo_rollout"

comp/core/workloadmeta/collectors/internal/nvml/nvml.go

@@ -129,6 +129,14 @@ func (c *collector) fillNVMLAttributes(gpuDeviceInfo *workloadmeta.GPU, device d
 		gpuDeviceInfo.PCIBusID = pciBusIDFromNVMLInfo(pciInfo)
 	}
 
+	fabricInfo, err := physicalDevice.GetGpuFabricInfo()
+	if err == nil {
+		if clusterUUID, cliqueID, ok := fabricInfoToTags(fabricInfo); ok {
+			gpuDeviceInfo.FabricClusterUUID = clusterUUID
+			gpuDeviceInfo.FabricCliqueID = cliqueID
+		}
+	}
+
 	// Do not generate errors for vGPU devices, we already know that they don't support max clock info
 	if virtMode != nvml.GPU_VIRTUALIZATION_MODE_VGPU {
 		maxSMClock, err := physicalDevice.GetMaxClockInfo(nvml.CLOCK_SM)
@@ -163,6 +171,20 @@ func pciBusIDFromNVMLInfo(pciInfo nvml.PciInfo) string {
 	return strings.ToLower(fmt.Sprintf("%04x:%02x:%02x.0", pciInfo.Domain, pciInfo.Bus, pciInfo.Device))
 }
 
+func fabricClusterUUIDFromNVMLInfo(clusterUUID [16]uint8) string {
+	return fmt.Sprintf("%x-%x-%x-%x-%x", clusterUUID[0:4], clusterUUID[4:6], clusterUUID[6:8], clusterUUID[8:10], clusterUUID[10:16])
+}
+
+func fabricInfoToTags(fabricInfo nvml.GpuFabricInfo_v2) (string, uint32, bool) {
+	if fabricInfo.State != nvml.GPU_FABRIC_STATE_COMPLETED ||
+		nvml.Return(fabricInfo.Status) != nvml.SUCCESS ||
+		fabricInfo.ClusterUuid == [16]uint8{} {
+		return "", 0, false
+	}
+
+	return fabricClusterUUIDFromNVMLInfo(fabricInfo.ClusterUuid), fabricInfo.CliqueId, true
+}
+
 func (c *collector) fillProcesses(gpuDeviceInfo *workloadmeta.GPU, device ddnvml.Device) {
 	seenPIDs := make(map[int]struct{})
 	procs, err := device.GetComputeRunningProcesses()

comp/core/workloadmeta/collectors/internal/nvml/nvml_test.go

@@ -64,6 +64,7 @@ func TestPull(t *testing.T) {
 		require.ElementsMatch(t, expectedGPUActivePIDs, gpu.ActivePIDs)
 		require.Equal(t, "none", gpu.VirtualizationMode)
 		require.Equal(t, "0000:00:1e.0", gpu.PCIBusID)
+		require.Empty(t, gpu.FabricClusterUUID)
 	}
 
 	for _, uuid := range testutil.GPUUUIDs {
@@ -77,6 +78,71 @@ func TestPull(t *testing.T) {
 	}
 }
 
+func TestFabricInfoToTags(t *testing.T) {
+	clusterUUID := [16]uint8{0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff}
+	tests := []struct {
+		name              string
+		fabricInfo        nvml.GpuFabricInfo_v2
+		expectedClusterID string
+		expectedCliqueID  uint32
+		expectedAvailable bool
+	}{
+		{
+			name: "completed fabric with cluster UUID",
+			fabricInfo: nvml.GpuFabricInfo_v2{
+				State:       nvml.GPU_FABRIC_STATE_COMPLETED,
+				Status:      uint32(nvml.SUCCESS),
+				CliqueId:    42,
+				ClusterUuid: clusterUUID,
+			},
+			expectedClusterID: "00112233-4455-6677-8899-aabbccddeeff",
+			expectedCliqueID:  42,
+			expectedAvailable: true,
+		},
+		{
+			name: "fabric initialization incomplete",
+			fabricInfo: nvml.GpuFabricInfo_v2{
+				State:       nvml.GPU_FABRIC_STATE_IN_PROGRESS,
+				Status:      uint32(nvml.SUCCESS),
+				CliqueId:    42,
+				ClusterUuid: clusterUUID,
+			},
+		},
+		{
+			name: "fabric status failed",
+			fabricInfo: nvml.GpuFabricInfo_v2{
+				State:       nvml.GPU_FABRIC_STATE_COMPLETED,
+				Status:      uint32(nvml.ERROR_UNKNOWN),
+				CliqueId:    42,
+				ClusterUuid: clusterUUID,
+			},
+		},
+		{
+			name: "cluster UUID is unavailable",
+			fabricInfo: nvml.GpuFabricInfo_v2{
+				State:    nvml.GPU_FABRIC_STATE_COMPLETED,
+				Status:   uint32(nvml.SUCCESS),
+				CliqueId: 42,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			clusterID, cliqueID, available := fabricInfoToTags(tt.fabricInfo)
+			require.Equal(t, tt.expectedClusterID, clusterID)
+			require.Equal(t, tt.expectedCliqueID, cliqueID)
+			require.Equal(t, tt.expectedAvailable, available)
+		})
+	}
+}
+
+func TestFabricClusterUUIDFromNVMLInfo(t *testing.T) {
+	clusterUUID := [16]uint8{0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff}
+
+	require.Equal(t, "00112233-4455-6677-8899-aabbccddeeff", fabricClusterUUIDFromNVMLInfo(clusterUUID))
+}
+
 func TestPCIBusIDFromNVMLInfo(t *testing.T) {
 	tests := []struct {
 		name     string

comp/core/workloadmeta/def/types.go

@@ -2315,6 +2315,14 @@ type GPU struct {
 	// PCIBusID is the PCI bus ID of the GPU in domain:bus:device.function format.
 	PCIBusID string
 
+	// FabricClusterUUID identifies the NVLink fabric cluster that contains the GPU.
+	// Empty when the GPU is not registered with a fabric cluster.
+	FabricClusterUUID string
+
+	// FabricCliqueID identifies the P2P clique within the NVLink fabric cluster.
+	// It is meaningful only when FabricClusterUUID is set.
+	FabricCliqueID uint32
+
 	// DeviceType identifies if this is a physical or virtual device (e.g. MIG)
 	DeviceType GPUDeviceType
 

pkg/gpu/safenvml/device.go

@@ -52,6 +52,8 @@ type SafeDevice interface {
 	GetGpuInstanceId() (int, error)
 	// GetGpuInstanceProfileInfo returns the profile info for the given GPU instance profile ID
 	GetGpuInstanceProfileInfo(profile int) (nvml.GpuInstanceProfileInfo, error)
+	// GetGpuFabricInfo returns the NVLink fabric information for the device.
+	GetGpuFabricInfo() (nvml.GpuFabricInfo_v2, error)
 	// GetIndex returns the index of the device
 	GetIndex() (int, error)
 	// GetMaxClockInfo returns the maximum clock speed for the given clock type

pkg/gpu/safenvml/device_impl.go

@@ -139,6 +139,17 @@ func (d *safeDeviceImpl) GetGpuInstanceProfileInfo(profile int) (nvml.GpuInstanc
 	return info, NewNvmlAPIErrorOrNil("GetGpuInstanceProfileInfo", ret)
 }
 
+func (d *safeDeviceImpl) GetGpuFabricInfo() (nvml.GpuFabricInfo_v2, error) {
+	if err := d.lib.lookup(toNativeName("GetGpuFabricInfoV")); err != nil {
+		return nvml.GpuFabricInfo_v2{}, err
+	}
+	info, ret := d.nvmlDevice.GetGpuFabricInfoV().V2()
+	if err := NewNvmlAPIErrorOrNil("GetGpuFabricInfoV", ret); err != nil {
+		return nvml.GpuFabricInfo_v2{}, err
+	}
+	return info, nil
+}
+
 func (d *safeDeviceImpl) GetIndex() (int, error) {
 	if err := d.lib.lookup(toNativeName("GetIndex")); err != nil {
 		return 0, err

pkg/gpu/safenvml/device_impl_test.go

@@ -141,3 +141,16 @@ func TestDeviceSafeMethodSuccess(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, testutil.DefaultGpuCores, cores)
 }
+
+func TestGetGpuFabricInfoRequiresVersionedAPISymbol(t *testing.T) {
+	symbols := maps.Clone(allSymbols)
+	delete(symbols, toNativeName("GetGpuFabricInfoV"))
+
+	device := &safeDeviceImpl{
+		lib: &safeNvml{capabilities: symbols},
+	}
+
+	_, err := device.GetGpuFabricInfo()
+	require.Error(t, err)
+	require.True(t, IsUnsupported(err))
+}

pkg/gpu/safenvml/lib.go

@@ -71,6 +71,7 @@ func getNonCriticalAPIs() []string {
 		toNativeName("GetFanSpeed_v2"),
 		toNativeName("GetFieldValues"),
 		"nvmlDeviceReadWritePRM_v1",
+		toNativeName("GetGpuFabricInfoV"),
 		toNativeName("GetGpuInstanceId"),
 		toNativeName("GetGpuInstanceProfileInfo"),
 		toNativeName("GetMaxClockInfo"),