feat(sysadvisor): support numa aware memory headroom policy

pkg/agent/sysadvisor/plugin/qosaware/resource/memory/advisor.go

@@ -41,6 +41,7 @@ import (
 
 func init() {
 	headroompolicy.RegisterInitializer(types.MemoryHeadroomPolicyCanonical, headroompolicy.NewPolicyCanonical)
+	headroompolicy.RegisterInitializer(types.MemoryHeadroomPolicyNUMAAware, headroompolicy.NewPolicyNUMAAware)
 
 	memadvisorplugin.RegisterInitializer(memadvisorplugin.CacheReaper, memadvisorplugin.NewCacheReaper)
 	memadvisorplugin.RegisterInitializer(memadvisorplugin.MemoryGuard, memadvisorplugin.NewMemoryGuard)
@@ -98,7 +99,10 @@ func NewMemoryResourceAdvisor(conf *config.Configuration, extraConf interface{},
 			klog.Errorf("failed to find registered initializer %v", headroomPolicyName)
 			continue
 		}
-		ra.headroomPolices = append(ra.headroomPolices, initFunc(conf, extraConf, metaCache, metaServer, emitter))
+		policy := initFunc(conf, extraConf, metaCache, metaServer, emitter)
+		general.InfoS("add new memory headroom policy", "policyName", policy.Name())
+
+		ra.headroomPolices = append(ra.headroomPolices, policy)
 	}
 
 	memoryAdvisorPluginInitializers := memadvisorplugin.GetRegisteredInitializers()
@@ -108,6 +112,7 @@ func NewMemoryResourceAdvisor(conf *config.Configuration, extraConf interface{},
 			klog.Errorf("failed to find registered initializer %v", memadvisorPluginName)
 			continue
 		}
+		general.InfoS("add new memory advisor policy", "policyName", memadvisorPluginName)
 		ra.plugins = append(ra.plugins, initFunc(conf, extraConf, metaCache, metaServer, emitter))
 	}
 
@@ -135,7 +140,7 @@ func (ra *memoryResourceAdvisor) GetHeadroom() (resource.Quantity, error) {
 	for _, headroomPolicy := range ra.headroomPolices {
 		headroom, err := headroomPolicy.GetHeadroom()
 		if err != nil {
-			klog.Warningf("[qosaware-memory] get headroom with error: %v", err)
+			klog.ErrorS(err, "get headroom failed", "headroomPolicy", headroomPolicy.Name())
 			continue
 		}
 		return headroom, nil
@@ -182,7 +187,7 @@ func (ra *memoryResourceAdvisor) update() {
 		})
 
 		if err := headroomPolicy.Update(); err != nil {
-			klog.Errorf("[qosaware-memory] update headroom policy failed: %v", err)
+			general.ErrorS(err, "[qosaware-memory] update headroom policy failed", "headroomPolicy", headroomPolicy.Name())
 		}
 	}
 

pkg/agent/sysadvisor/plugin/qosaware/resource/memory/headroompolicy/policy.go

@@ -30,6 +30,7 @@ import (
 
 // HeadroomPolicy generates resource headroom estimation based on configured algorithm
 type HeadroomPolicy interface {
+	Name() types.MemoryHeadroomPolicyName
 	// SetPodSet overwrites policy's pod/container record
 	SetPodSet(types.PodSet)
 

pkg/agent/sysadvisor/plugin/qosaware/resource/memory/headroompolicy/policy_canonical.go

@@ -56,6 +56,10 @@ func NewPolicyCanonical(conf *config.Configuration, _ interface{}, metaReader me
 	return &p
 }
 
+func (p *PolicyCanonical) Name() types.MemoryHeadroomPolicyName {
+	return types.MemoryHeadroomPolicyCanonical
+}
+
 // estimateNonReclaimedQoSMemoryRequirement estimates the memory requirement of all containers that are not reclaimed
 func (p *PolicyCanonical) estimateNonReclaimedQoSMemoryRequirement() (float64, error) {
 	var (

pkg/agent/sysadvisor/plugin/qosaware/resource/memory/headroompolicy/policy_numa_aware.go

@@ -0,0 +1,155 @@
+/*
+Copyright 2022 The Katalyst Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package headroompolicy
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/errors"
+
+	apiconsts "github.com/kubewharf/katalyst-api/pkg/consts"
+	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/metacache"
+	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/qosaware/resource/helper"
+	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/types"
+	"github.com/kubewharf/katalyst-core/pkg/config"
+	"github.com/kubewharf/katalyst-core/pkg/consts"
+	"github.com/kubewharf/katalyst-core/pkg/metaserver"
+	"github.com/kubewharf/katalyst-core/pkg/metrics"
+	"github.com/kubewharf/katalyst-core/pkg/util/general"
+	"github.com/kubewharf/katalyst-core/pkg/util/machine"
+	"github.com/kubewharf/katalyst-core/pkg/util/metric"
+)
+
+type PolicyNUMAAware struct {
+	*PolicyBase
+
+	// memoryHeadroom is valid to be used iff updateStatus successes
+	memoryHeadroom float64
+	updateStatus   types.PolicyUpdateStatus
+
+	conf *config.Configuration
+}
+
+func NewPolicyNUMAAware(conf *config.Configuration, _ interface{}, metaReader metacache.MetaReader,
+	metaServer *metaserver.MetaServer, _ metrics.MetricEmitter) HeadroomPolicy {
+	p := PolicyNUMAAware{
+		PolicyBase:   NewPolicyBase(metaReader, metaServer),
+		updateStatus: types.PolicyUpdateFailed,
+		conf:         conf,
+	}
+
+	return &p
+}
+
+func (p *PolicyNUMAAware) Name() types.MemoryHeadroomPolicyName {
+	return types.MemoryHeadroomPolicyNUMAAware
+}
+
+func (p *PolicyNUMAAware) reclaimedContainersFilter(ci *types.ContainerInfo) bool {
+	return ci != nil && ci.QoSLevel == apiconsts.PodAnnotationQoSLevelReclaimedCores
+}
+
+func (p *PolicyNUMAAware) Update() (err error) {
+	defer func() {
+		if err != nil {
+			p.updateStatus = types.PolicyUpdateFailed
+		} else {
+			p.updateStatus = types.PolicyUpdateSucceeded
+		}
+	}()
+
+	var (
+		errList           []error
+		reclaimableMemory float64
+		data              metric.MetricData
+	)
+
+	availNUMAs := p.metaServer.CPUDetails.NUMANodes()
+
+	reclaimedCoresContainers := make([]*types.ContainerInfo, 0)
+	p.metaReader.RangeContainer(func(podUID string, containerName string, containerInfo *types.ContainerInfo) bool {
+		if p.reclaimedContainersFilter(containerInfo) {
+			reclaimedCoresContainers = append(reclaimedCoresContainers, containerInfo)
+			return true
+		}
+
+		nodeReclaim := p.conf.GetDynamicConfiguration().EnableReclaim
+		reclaimEnable, err := helper.PodEnableReclaim(context.Background(), p.metaServer, podUID, nodeReclaim)
+		if err != nil {
+			errList = append(errList, err)
+			return true
+		}
+
+		if containerInfo.IsNumaExclusive() && !reclaimEnable {
+			memset := machine.GetCPUAssignmentNUMAs(containerInfo.TopologyAwareAssignments)
+			if memset.IsEmpty() {
+				errList = append(errList, fmt.Errorf("container(%v/%v) TopologyAwareAssignments is empty", containerInfo.PodName, containerName))
+				return true
+			}
+			availNUMAs = availNUMAs.Difference(memset)
+		}
+		return true
+	})
+
+	err = errors.NewAggregate(errList)
+	if err != nil {
+		return err
+	}
+
+	for _, numaID := range availNUMAs.ToSliceInt() {
+		data, err = p.metaServer.GetNumaMetric(numaID, consts.MetricMemFreeNuma)
+		if err != nil {
+			return err
+		}
+		numaFree := data.Value
+
+		reclaimedCoresUsed := 0.
+		for _, container := range reclaimedCoresContainers {
+			data, err = p.metaServer.GetContainerNumaMetric(container.PodUID, container.ContainerName, strconv.Itoa(numaID), consts.MetricsMemTotalPerNumaContainer)
+			if err != nil {
+				general.ErrorS(err, "failed to get metric", "name", consts.MetricsMemTotalPerNumaContainer,
+					"podName", container.PodName, "ContainerName", container.ContainerName, "numaID", numaID)
+				return err
+			}
+			reclaimedCoresUsed += data.Value
+			general.InfoS("container memory", "podName", container.PodName, "containerName", container.ContainerName, "numaID", numaID, "memory used", general.FormatMemoryQuantity(data.Value))
+		}
+
+		reclaimableMemory += numaFree + reclaimedCoresUsed
+
+		general.InfoS("memory reclaimable", "numaID", numaID, "numaFree", general.FormatMemoryQuantity(numaFree), "reclaimable", general.FormatMemoryQuantity(numaFree+reclaimedCoresUsed))
+	}
+
+	general.InfoS("total memory reclaimable",
+		"reclaimableMemory", general.FormatMemoryQuantity(reclaimableMemory),
+		"ReservedForAllocate", general.FormatMemoryQuantity(p.essentials.ReservedForAllocate),
+		"ResourceUpperBound", general.FormatMemoryQuantity(p.essentials.ResourceUpperBound))
+	p.memoryHeadroom = general.Clamp(reclaimableMemory-p.essentials.ReservedForAllocate, 0, p.essentials.ResourceUpperBound)
+
+	return nil
+}
+
+func (p *PolicyNUMAAware) GetHeadroom() (resource.Quantity, error) {
+	if p.updateStatus != types.PolicyUpdateSucceeded {
+		return resource.Quantity{}, fmt.Errorf("last update failed")
+	}
+
+	return *resource.NewQuantity(int64(p.memoryHeadroom), resource.BinarySI), nil
+}