feat(sysadvisor): report numa memory_bandwidth to kcnr

Author: syc4704413

go.mod

@@ -176,6 +176,7 @@ require (
 )
 
 replace (
+	github.com/kubewharf/katalyst-api => github.com/syc4704413/katalyst-api v0.0.2-syc-mem-bw
 	k8s.io/api => k8s.io/api v0.24.6
 	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.24.6
 	k8s.io/apimachinery => k8s.io/apimachinery v0.24.6

go.sum

@@ -574,8 +574,6 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/kubewharf/katalyst-api v0.5.11-0.20260423040236-f1a2330d266e h1:/0LP1rzsXEI09g3eqUZCJcyv/XGU8MofdFyi7NBtZA0=
-github.com/kubewharf/katalyst-api v0.5.11-0.20260423040236-f1a2330d266e/go.mod h1:BZMVGVl3EP0eCn5xsDgV41/gjYkoh43abIYxrB10e3k=
 github.com/kubewharf/kubelet v1.24.6-kubewharf-pre.3 h1:oFwpVVeDESqgzkse3iSODzHRKX495VvUgslu2hhepCc=
 github.com/kubewharf/kubelet v1.24.6-kubewharf-pre.3/go.mod h1:MxbSZUx3wXztFneeelwWWlX7NAAStJ6expqq7gY2J3c=
 github.com/kyoh86/exportloopref v0.1.7/go.mod h1:h1rDl2Kdj97+Kwh4gdz3ujE7XHmH51Q0lUiZ1z4NLj8=
@@ -900,6 +898,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.3 h1:RP3t2pwF7cMEbC1dqtB6poj3niw/9gnV4Cjg5oW5gtY=
 github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
+github.com/syc4704413/katalyst-api v0.0.2-syc-mem-bw h1:z70YV8teGYYdXFTP19Ayrp5vpT2uxwF6ihjH3gNf4p0=
+github.com/syc4704413/katalyst-api v0.0.2-syc-mem-bw/go.mod h1:BZMVGVl3EP0eCn5xsDgV41/gjYkoh43abIYxrB10e3k=
 github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635 h1:kdXcSzyDtseVEc4yCz2qF8ZrQvIDBJLl4S1c3GCXmoI=
 github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww=
 github.com/tdakkota/asciicheck v0.0.0-20200416190851-d7f85be797a2/go.mod h1:yHp0ai0Z9gUljN3o0xMhYJnH/IcvkdTBOX2fmJ93JEM=

pkg/agent/sysadvisor/plugin/qosaware/reporter/nodemetric_reporter.go

@@ -43,6 +43,7 @@ import (
 	"github.com/kubewharf/katalyst-core/pkg/config/generic"
 	"github.com/kubewharf/katalyst-core/pkg/consts"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver"
+	malachitetypes "github.com/kubewharf/katalyst-core/pkg/metaserver/agent/metric/provisioner/malachite/types"
 	"github.com/kubewharf/katalyst-core/pkg/metrics"
 	schedutil "github.com/kubewharf/katalyst-core/pkg/scheduler/util"
 	"github.com/kubewharf/katalyst-core/pkg/util"
@@ -303,6 +304,13 @@ func (p *nodeMetricsReporterPlugin) getNodeMetricInfo() (*nodeapis.NodeMetricInf
 		nmi.GenericUsage.CPU = cpuUsage
 	}
 
+	memoryBandwidthUsage, err := p.getNodeMemoryBandwidthUsage()
+	if err != nil {
+		errList = append(errList, err)
+	} else {
+		nmi.GenericUsage.MemoryBandwidth = memoryBandwidthUsage
+	}
+
 	for numaID := 0; numaID < p.metaServer.NumNUMANodes; numaID++ {
 		numaUsage := nodeapis.NUMAMetricInfo{NUMAId: numaID, Usage: &nodeapis.ResourceMetric{}}
 		memoryUsage, err := p.getNodeNUMAMemoryUsage(numaID)
@@ -319,6 +327,13 @@ func (p *nodeMetricsReporterPlugin) getNodeMetricInfo() (*nodeapis.NodeMetricInf
 			numaUsage.Usage.CPU = numaCpuUsage
 		}
 
+		numaMemoryBandwidthUsage, err := p.getNodeNUMAMemoryBandwidthUsage(numaID)
+		if err != nil {
+			errList = append(errList, err)
+		} else {
+			numaUsage.Usage.MemoryBandwidth = numaMemoryBandwidthUsage
+		}
+
 		nmi.NUMAUsage = append(nmi.NUMAUsage, numaUsage)
 	}
 	return nmi, errors.NewAggregate(errList)
@@ -394,6 +409,7 @@ func (p *nodeMetricsReporterPlugin) getPodUsage(pod *v1.Pod) (v1.ResourceList, m
 	}
 	podCPUUsage := .0
 	podMemUsage := .0
+	podMemoryBandwidthUsage := resource.NewQuantity(0, resource.BinarySI)
 	for _, container := range containers {
 		if container.RampUp {
 			rampUp = true
@@ -460,19 +476,46 @@ func (p *nodeMetricsReporterPlugin) getPodUsage(pod *v1.Pod) (v1.ResourceList, m
 			usages[v1.ResourceMemory] = memUsage
 			numaUsage[numaID] = usages
 		}
+
+		containerNUMAMBWUsage, containerTotalMBW, err := p.getContainerNUMAMemoryBandwidthUsage(string(pod.UID), container.ContainerName)
+		if err != nil {
+			errList = append(errList, fmt.Errorf("failed to get container NUMA memory bandwidth usage, podUID=%v, containerName=%v, err=%v",
+				pod.UID, container.ContainerName, err))
+		} else {
+			podMemoryBandwidthUsage.Add(*containerTotalMBW)
+			for numaID, mbw := range containerNUMAMBWUsage {
+				usages, ok := numaUsage[numaID]
+				if !ok {
+					usages = make(v1.ResourceList)
+				}
+
+				mbwUsage, ok := usages[apiconsts.ResourceMemoryBandwidth]
+				if !ok {
+					mbwUsage = *resource.NewQuantity(0, resource.BinarySI)
+				}
+				mbwUsage.Add(mbw)
+				usages[apiconsts.ResourceMemoryBandwidth] = mbwUsage
+				numaUsage[numaID] = usages
+			}
+		}
 	}
 
 	cpu := resource.NewMilliQuantity(int64(podCPUUsage*1000), resource.DecimalSI)
 	memory := resource.NewQuantity(int64(podMemUsage), resource.BinarySI)
 
-	return v1.ResourceList{v1.ResourceMemory: *memory, v1.ResourceCPU: *cpu}, numaUsage, assignedNUMAs, rampUp, errors.NewAggregate(errList)
+	return v1.ResourceList{
+		v1.ResourceMemory:                 *memory,
+		v1.ResourceCPU:                    *cpu,
+		apiconsts.ResourceMemoryBandwidth: *podMemoryBandwidthUsage,
+	}, numaUsage, assignedNUMAs, rampUp, errors.NewAggregate(errList)
 }
 
 func (p *nodeMetricsReporterPlugin) getGroupUsage(pods []*v1.Pod, qosLevel string) (*nodeapis.ResourceMetric, []nodeapis.NUMAMetricInfo, []*v1.Pod, error) {
 	var errList []error
 
 	cpu := resource.NewQuantity(0, resource.DecimalSI)
 	memory := resource.NewQuantity(0, resource.BinarySI)
+	memoryBandwidth := resource.NewQuantity(0, resource.BinarySI)
 
 	numaUsages := make(map[int]v1.ResourceList)
 
@@ -494,14 +537,28 @@ func (p *nodeMetricsReporterPlugin) getGroupUsage(pods []*v1.Pod, qosLevel strin
 			}
 
 			for numaID := range assignedNUMAs.ToSliceInt() {
-				podNUMAUsage[numaID] = map[v1.ResourceName]resource.Quantity{
-					v1.ResourceCPU:    *resource.NewMilliQuantity(req.Cpu().MilliValue()/int64(assignedNUMAs.Size()), resource.DecimalSI),
-					v1.ResourceMemory: *resource.NewQuantity(req.Memory().Value()/int64(assignedNUMAs.Size()), resource.BinarySI),
+				usages, ok := podNUMAUsage[numaID]
+				if !ok {
+					usages = make(v1.ResourceList)
 				}
+
+				usages[v1.ResourceCPU] = *resource.NewMilliQuantity(
+					req.Cpu().MilliValue()/int64(assignedNUMAs.Size()),
+					resource.DecimalSI,
+				)
+				usages[v1.ResourceMemory] = *resource.NewQuantity(
+					req.Memory().Value()/int64(assignedNUMAs.Size()),
+					resource.BinarySI,
+				)
+
+				podNUMAUsage[numaID] = usages
 			}
 		}
 		cpu.Add(*podUsage.Cpu())
 		memory.Add(*podUsage.Memory())
+		if mbw, ok := podUsage[apiconsts.ResourceMemoryBandwidth]; ok {
+			memoryBandwidth.Add(mbw)
+		}
 
 		for numaID, podUsages := range podNUMAUsage {
 			usages, ok := numaUsages[numaID]
@@ -539,6 +596,13 @@ func (p *nodeMetricsReporterPlugin) getGroupUsage(pods []*v1.Pod, qosLevel strin
 		resourceMetric.CPU = aggCPU
 	}
 
+	aggMBW := p.getAggregatedMetric(memoryBandwidth, apiconsts.ResourceMemoryBandwidth, "getGroupUsage", qosLevel, "memory_bandwidth")
+	if aggMBW == nil {
+		errList = append(errList, fmt.Errorf("failed to get enough samples for group memory bandwidth, qosLevel=%v", qosLevel))
+	} else {
+		resourceMetric.MemoryBandwidth = aggMBW
+	}
+
 	for numaID := 0; numaID < p.metaServer.NumNUMANodes; numaID++ {
 		resourceUsages, ok := numaUsages[numaID]
 		if !ok {
@@ -565,6 +629,14 @@ func (p *nodeMetricsReporterPlugin) getGroupUsage(pods []*v1.Pod, qosLevel strin
 			resourceNUMAMetric.Memory = aggNUMAMem
 		}
 
+		mbwUsage := resourceUsages[apiconsts.ResourceMemoryBandwidth]
+		aggNUMAMBW := p.getAggregatedMetric(&mbwUsage, apiconsts.ResourceMemoryBandwidth, "getGroupNUMAUsage", qosLevel, "memory_bandwidth", strconv.Itoa(numaID))
+		if aggNUMAMBW == nil {
+			errList = append(errList, fmt.Errorf("failed to get enough samples for group numa memory bandwidth, qosLevel=%v, numa=%v", qosLevel, numaID))
+		} else {
+			resourceNUMAMetric.MemoryBandwidth = aggNUMAMBW
+		}
+
 		resourceNUMAMetrics = append(resourceNUMAMetrics, nodeapis.NUMAMetricInfo{
 			NUMAId: numaID,
 			Usage:  &resourceNUMAMetric,
@@ -660,3 +732,52 @@ func (p *nodeMetricsReporterPlugin) getAggregatedMetric(value *resource.Quantity
 	}
 	return aggregator.GetWindowedResources(*value)
 }
+
+func (p *nodeMetricsReporterPlugin) getNodeNUMAMemoryBandwidthUsage(numaID int) (*resource.Quantity, error) {
+	metricData, err := p.metaServer.GetNumaMetric(numaID, consts.MetricTotalPsMemBandwidthNuma)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get %s for numa=%d: %w", consts.MetricTotalPsMemBandwidthNuma, numaID, err)
+	}
+
+	return resource.NewQuantity(int64(metricData.Value), resource.BinarySI), nil
+}
+
+func (p *nodeMetricsReporterPlugin) getNodeMemoryBandwidthUsage() (*resource.Quantity, error) {
+	total := resource.NewQuantity(0, resource.BinarySI)
+
+	for numaID := 0; numaID < p.metaServer.NumNUMANodes; numaID++ {
+		q, err := p.getNodeNUMAMemoryBandwidthUsage(numaID)
+		if err != nil {
+			return nil, err
+		}
+		total.Add(*q)
+	}
+	return total, nil
+}
+
+func (p *nodeMetricsReporterPlugin) getContainerNUMAMemoryBandwidthUsage(podUID, containerName string) (map[int]resource.Quantity, *resource.Quantity, error) {
+	key := fmt.Sprintf("%s/%s/%s", consts.MetricMbmTotalPsContainerL3, podUID, containerName)
+	raw := p.metaServer.MetricsFetcher.GetByStringIndex(key)
+
+	statsMap, ok := raw.(map[int]malachitetypes.L3CacheBytesPS)
+	if !ok || len(statsMap) == 0 {
+		return map[int]resource.Quantity{}, resource.NewQuantity(0, resource.BinarySI), nil
+	}
+
+	total := resource.NewQuantity(0, resource.BinarySI)
+	numaUsage := make(map[int]resource.Quantity)
+
+	for _, l3Stat := range statsMap {
+		q := *resource.NewQuantity(int64(l3Stat.MbmTotalBytesPS), resource.BinarySI)
+		total.Add(q)
+
+		exist, ok := numaUsage[l3Stat.NumaID]
+		if !ok {
+			exist = *resource.NewQuantity(0, resource.BinarySI)
+		}
+		exist.Add(q)
+		numaUsage[l3Stat.NumaID] = exist
+	}
+
+	return numaUsage, total, nil
+}

pkg/consts/metric.go

@@ -345,6 +345,8 @@ const (
 	MetricMbmlocalPsContainer  = "mbm.local.ps.container"
 	MetricMbmVictimPsContainer = "mbm.victim.ps.container"
 	MetricResctrlDataContainer = "resctrl.data.container"
+
+	MetricMbmTotalPsContainerL3 = "mbm.total.ps.container.l3"
 )
 
 // container blkio metrics

pkg/metaserver/agent/metric/provisioner/malachite/provisioner_calculate.go

@@ -338,19 +338,26 @@ func (m *MalachiteMetricsProvisioner) setContainerRateMetric(podUID, containerNa
 
 // setContainerMbmTotalMetric calcuate the total memory bandwidth usage of a container
 func (m *MalachiteMetricsProvisioner) setContainerMbmTotalMetric(podUID, containerName string, resctrlData types.MbmbandData, updateTime *time.Time) {
+	resctrlKey := getContainerStringIndexKey(consts.MetricResctrlDataContainer, podUID, containerName)
+	l3MetricKey := getContainerStringIndexKey(consts.MetricMbmTotalPsContainerL3, podUID, containerName)
+
 	// 1. get previous data
-	oldResctrlData, ok := m.metricStore.GetByStringIndex(consts.MetricResctrlDataContainer).(types.MbmbandData)
+	oldResctrlData, ok := m.metricStore.GetByStringIndex(resctrlKey).(types.MbmbandData)
 	if !ok || len(oldResctrlData.Mbm) == 0 || len(resctrlData.Mbm) == 0 {
 		// No previous or current data, store current for next round and return
-		m.metricStore.SetByStringIndex(consts.MetricResctrlDataContainer, resctrlData)
+		//m.metricStore.SetByStringIndex(consts.MetricResctrlDataContainer, resctrlData)
+		m.metricStore.SetByStringIndex(resctrlKey, resctrlData)
+		m.metricStore.SetByStringIndex(l3MetricKey, map[int]types.L3CacheBytesPS{})
 		return
 	}
 
 	// 2. calculate the time interval
 	prevMetric, err := m.metricStore.GetContainerMetric(podUID, containerName, consts.MetricMbmTotalPsContainer)
 	if err != nil || prevMetric.Time == nil {
 		// if no previous metric, store current for next round and return
-		m.metricStore.SetByStringIndex(consts.MetricResctrlDataContainer, resctrlData)
+		//m.metricStore.SetByStringIndex(consts.MetricResctrlDataContainer, resctrlData)
+		m.metricStore.SetByStringIndex(resctrlKey, resctrlData)
+		m.metricStore.SetByStringIndex(l3MetricKey, map[int]types.L3CacheBytesPS{})
 		m.metricStore.SetContainerMetric(podUID, containerName, consts.MetricMbmTotalPsContainer,
 			metric.MetricData{Value: 0, Time: updateTime})
 		m.metricStore.SetContainerMetric(podUID, containerName, consts.MetricMbmlocalPsContainer,
@@ -362,7 +369,15 @@ func (m *MalachiteMetricsProvisioner) setContainerMbmTotalMetric(podUID, contain
 	if timeInterval <= 0 {
 		return
 	}
+
+	cpuCodeName := ""
+	if val, ok := m.metricStore.GetByStringIndex(consts.MetricCPUCodeName).(string); ok {
+		cpuCodeName = val
+	}
+
 	var totalMbmBytesPS, totalLocalBytesPS float64
+	l3CacheBandwidthStats := make(map[int]types.L3CacheBytesPS)
+
 	for _, l3Mon := range resctrlData.Mbm {
 		l3CacheID := l3Mon.ID
 
@@ -395,16 +410,27 @@ func (m *MalachiteMetricsProvisioner) setContainerMbmTotalMetric(podUID, contain
 				continue
 			}
 		}
+
+		numaID, maxBytesPS := getNumaAndMaxBandwidth(int(l3CacheID), cpuCodeName)
+		adjustedTotalBytesPS := totalBytesPS
+
+		if strings.Contains(cpuCodeName, consts.AMDGenoaArch) {
+			// Notice: data adjustments needed due to genoa hardware bug
+			adjustedTotalBytesPS += localBytesPS / 3 * 2
+		}
+		totalMbmBytesPS += adjustedTotalBytesPS
 		totalLocalBytesPS += localBytesPS
-		cpuCodeName := m.metricStore.GetByStringIndex(consts.MetricCPUCodeName)
-		if codeName, ok := cpuCodeName.(string); ok {
-			if strings.Contains(codeName, consts.AMDGenoaArch) {
-				// Notice: data adjustments needed due to genoa hardware bug
-				totalLocalBytesPS += localBytesPS / 3 * 2
-			}
+		l3CacheBandwidthStats[int(l3CacheID)] = types.L3CacheBytesPS{
+			NumaID:           numaID,
+			MbmTotalBytesPS:  uint64(adjustedTotalBytesPS),
+			MbmLocalBytesPS:  uint64(localBytesPS),
+			MbmVictimBytesPS: 0,
+			MBMMaxBytesPS:    maxBytesPS,
 		}
 	}
-	m.metricStore.SetByStringIndex(consts.MetricResctrlDataContainer, resctrlData)
+	//m.metricStore.SetByStringIndex(consts.MetricResctrlDataContainer, resctrlData)
+	m.metricStore.SetByStringIndex(resctrlKey, resctrlData)
+	m.metricStore.SetByStringIndex(l3MetricKey, l3CacheBandwidthStats)
 	m.metricStore.SetContainerMetric(podUID, containerName, consts.MetricMbmTotalPsContainer,
 		metric.MetricData{Value: totalMbmBytesPS, Time: updateTime})
 	m.metricStore.SetContainerMetric(podUID, containerName, consts.MetricMbmlocalPsContainer,
@@ -581,3 +607,7 @@ func aggregateNUMABytesPS(l3BytesPS map[int]types.L3CacheBytesPS, cpuCodeName st
 	}
 	return numaBytesPS
 }
+
+func getContainerStringIndexKey(metricName, podUID, containerName string) string {
+	return fmt.Sprintf("%s/%s/%s", metricName, podUID, containerName)
+}