feat: introduce memset binder

Signed-off-by: linzhecheng <linzhecheng@bytedance.com>

Author: cheney-lin

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

@@ -45,6 +45,7 @@ func init() {
 	headroompolicy.RegisterInitializer(types.MemoryHeadroomPolicyCanonical, headroompolicy.NewPolicyCanonical)
 	memadvisorplugin.RegisterInitializer(memadvisorplugin.CacheReaper, memadvisorplugin.NewCacheReaper)
 	memadvisorplugin.RegisterInitializer(memadvisorplugin.MemoryGuard, memadvisorplugin.NewMemoryGuard)
+	memadvisorplugin.RegisterInitializer(memadvisorplugin.MemsetBinder, memadvisorplugin.NewMemsetBinder)
 }
 
 const (
@@ -230,6 +231,10 @@ func (ra *memoryResourceAdvisor) detectNUMAPressure(numaID int) (*types.MemoryPr
 		return nil, err
 	}
 
+	general.Infof("numa %v metrics, free: %+v, total: %+v, scaleFactor: %+v", numaID,
+		resource.NewQuantity(int64(free), resource.BinarySI).String(),
+		resource.NewQuantity(int64(total), resource.BinarySI).String(), scaleFactor)
+
 	targetReclaimed := resource.NewQuantity(0, resource.BinarySI)
 
 	pressureState := types.MemoryPressureNoRisk
@@ -253,7 +258,9 @@ func (ra *memoryResourceAdvisor) detectNodePressureCondition() (*types.MemoryPre
 		return nil, err
 	}
 
-	general.Infof("system watermark metrics, free: %+v, total: %+v, scaleFactor: %+v", free, total, scaleFactor)
+	general.Infof("system watermark metrics, free: %+v, total: %+v, scaleFactor: %+v",
+		resource.NewQuantity(int64(free), resource.BinarySI).String(),
+		resource.NewQuantity(int64(total), resource.BinarySI).String(), scaleFactor)
 
 	targetReclaimed := resource.NewQuantity(0, resource.BinarySI)
 

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

@@ -234,21 +234,6 @@ var defaultNumaMetrics = []numaMetric{
 	},
 }
 
-var tuneMemcgNodeMetrics = []nodeMetric{
-	{
-		metricName:  coreconsts.MetricMemFreeSystem,
-		metricValue: metricutil.MetricData{Value: 80 << 30},
-	},
-	{
-		metricName:  coreconsts.MetricMemTotalSystem,
-		metricValue: metricutil.MetricData{Value: 500 << 30},
-	},
-	{
-		metricName:  coreconsts.MetricMemScaleFactorSystem,
-		metricValue: metricutil.MetricData{Value: 500},
-	},
-}
-
 var dropCacheNodeMetrics = []nodeMetric{
 	{
 		metricName:  coreconsts.MetricMemFreeSystem,
@@ -634,6 +619,63 @@ func TestUpdate(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "bind memset",
+			pools: map[string]*types.PoolInfo{
+				state.PoolNameReserve: {
+					PoolName: state.PoolNameReserve,
+					TopologyAwareAssignments: map[int]machine.CPUSet{
+						0: machine.MustParse("0"),
+						1: machine.MustParse("24"),
+					},
+					OriginalTopologyAwareAssignments: map[int]machine.CPUSet{
+						0: machine.MustParse("0"),
+						1: machine.MustParse("24"),
+					},
+				},
+			},
+			reclaimedEnable: false,
+			needRecvAdvices: true,
+			containers: []*types.ContainerInfo{
+				makeContainerInfo("uid1", "default", "pod1", "c1", consts.PodAnnotationQoSLevelReclaimedCores, nil,
+					map[int]machine.CPUSet{
+						0: machine.MustParse("1"),
+						1: machine.MustParse("25"),
+					}, 200<<30),
+				makeContainerInfo("uid2", "default", "pod2", "c2", consts.PodAnnotationQoSLevelReclaimedCores, nil,
+					map[int]machine.CPUSet{
+						0: machine.MustParse("1"),
+						1: machine.MustParse("25"),
+					}, 200<<30),
+				makeContainerInfo("uid3", "default", "pod3", "c3", consts.PodAnnotationQoSLevelReclaimedCores, nil,
+					map[int]machine.CPUSet{
+						0: machine.MustParse("1"),
+					}, 200<<30),
+			},
+			plugins:      []types.MemoryAdvisorPluginName{memadvisorplugin.MemsetBinder},
+			nodeMetrics:  defaultNodeMetrics,
+			numaMetrics:  defaultNumaMetrics,
+			wantHeadroom: *resource.NewQuantity(996<<30, resource.DecimalSI),
+			wantAdviceResult: types.InternalMemoryCalculationResult{
+				ContainerEntries: []types.ContainerMemoryAdvices{
+					{
+						PodUID:        "uid1",
+						ContainerName: "c1",
+						Values:        map[string]string{string(memoryadvisor.ControKnobKeyCPUSetMems): "0-1"},
+					},
+					{
+						PodUID:        "uid2",
+						ContainerName: "c2",
+						Values:        map[string]string{string(memoryadvisor.ControKnobKeyCPUSetMems): "0-1"},
+					},
+					{
+						PodUID:        "uid3",
+						ContainerName: "c3",
+						Values:        map[string]string{string(memoryadvisor.ControKnobKeyCPUSetMems): "0"},
+					},
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {
@@ -687,8 +729,8 @@ func TestUpdate(t *testing.T) {
 			if tt.needRecvAdvices {
 				result := <-recvCh
 
-				assert.Equal(t, tt.wantAdviceResult.ExtraEntries, result.ExtraEntries)
-				assert.Equal(t, tt.wantAdviceResult.ContainerEntries, result.ContainerEntries)
+				assert.ElementsMatch(t, tt.wantAdviceResult.ExtraEntries, result.ExtraEntries)
+				assert.ElementsMatch(t, tt.wantAdviceResult.ContainerEntries, result.ContainerEntries)
 			}
 			headroom, err := advisor.GetHeadroom()
 

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

@@ -0,0 +1,90 @@
+/*
+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 plugin
+
+import (
+	"sync"
+
+	apiconsts "github.com/kubewharf/katalyst-api/pkg/consts"
+	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/memory/dynamicpolicy/memoryadvisor"
+	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/metacache"
+	"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/native"
+)
+
+const (
+	MemsetBinder = "memset-binder"
+)
+
+type memsetBinder struct {
+	mutex           sync.RWMutex
+	metaReader      metacache.MetaReader
+	emitter         metrics.MetricEmitter
+	containerMemset map[consts.PodContainerName]machine.CPUSet
+}
+
+func NewMemsetBinder(conf *config.Configuration, extraConfig interface{}, metaReader metacache.MetaReader, metaServer *metaserver.MetaServer, emitter metrics.MetricEmitter) MemoryAdvisorPlugin {
+	return &memsetBinder{
+		metaReader: metaReader,
+		emitter:    emitter,
+	}
+}
+
+func (mb *memsetBinder) reclaimedContainersFilter(ci *types.ContainerInfo) bool {
+	return ci != nil && ci.QoSLevel == apiconsts.PodAnnotationQoSLevelReclaimedCores
+}
+
+func (mb *memsetBinder) Reconcile(status *types.MemoryPressureStatus) error {
+	containerMemset := make(map[consts.PodContainerName]machine.CPUSet)
+	containers := mb.metaReader.GetContainers(mb.reclaimedContainersFilter)
+	for _, ci := range containers {
+		memset := machine.CPUSet2MemSet(ci.TopologyAwareAssignments)
+		containerMemset[native.GeneratePodContainerName(ci.PodUID, ci.ContainerName)] = memset
+	}
+	mb.mutex.Lock()
+	defer mb.mutex.Unlock()
+	mb.containerMemset = containerMemset
+
+	return nil
+}
+
+func (mb *memsetBinder) GetAdvices() types.InternalMemoryCalculationResult {
+	mb.mutex.RLock()
+	defer mb.mutex.RUnlock()
+	result := types.InternalMemoryCalculationResult{}
+	for podContainerName, memset := range mb.containerMemset {
+		podUID, containerName, err := native.ParsePodContainerName(podContainerName)
+		if err != nil {
+			general.Errorf("parse podContainerName %v err %v", podContainerName, err)
+			continue
+		}
+		entry := types.ContainerMemoryAdvices{
+			PodUID:        podUID,
+			ContainerName: containerName,
+			Values:        map[string]string{string(memoryadvisor.ControKnobKeyCPUSetMems): memset.String()},
+		}
+		result.ContainerEntries = append(result.ContainerEntries, entry)
+	}
+
+	return result
+}

pkg/util/machine/util.go

@@ -110,3 +110,14 @@ func GetSiblingNUMAs(numaID int, topology *CPUTopology) (CPUSet, error) {
 
 	return numaSet, nil
 }
+
+// CPUSet2MemSet returns memset for cpuset
+func CPUSet2MemSet(assignment map[int]CPUSet) CPUSet {
+	memset := NewCPUSet()
+	for numaID, cpuset := range assignment {
+		if cpuset.Size() > 0 {
+			memset.Add(numaID)
+		}
+	}
+	return memset
+}