feat: adding support to qos level memory.low protection

cmd/katalyst-agent/app/options/qrm/memory_plugin.go

@@ -33,6 +33,7 @@ type MemoryOptions struct {
 	OOMPriorityPinnedMapAbsPath string
 
 	SockMemOptions
+	MemLow
 }
 
 type SockMemOptions struct {
@@ -43,6 +44,11 @@ type SockMemOptions struct {
 	SetCgroupTCPMemRatio int
 }
 
+type MemLow struct {
+	EnableSettingMemLow      bool
+	MemLowQoSLevelConfigFile string
+}
+
 func NewMemoryOptions() *MemoryOptions {
 	return &MemoryOptions{
 		PolicyName:                 "dynamic",
@@ -56,6 +62,10 @@ func NewMemoryOptions() *MemoryOptions {
 			SetGlobalTCPMemRatio: 20,  // default: 20% * {host total memory}
 			SetCgroupTCPMemRatio: 100, // default: 100% * {cgroup memory}
 		},
+		MemLow: MemLow{
+			EnableSettingMemLow:      false,
+			MemLowQoSLevelConfigFile: "",
+		},
 	}
 }
 
@@ -84,6 +94,10 @@ func (o *MemoryOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		o.SetGlobalTCPMemRatio, "limit global max tcp memory usage")
 	fs.IntVar(&o.SetCgroupTCPMemRatio, "qrm-memory-cgroup-tcpmem-ratio",
 		o.SetCgroupTCPMemRatio, "limit cgroup max tcp memory usage")
+	fs.BoolVar(&o.EnableSettingMemLow, "enable-setting-mem-low",
+		o.EnableSettingMemLow, "if set true, we will do memory soft protection in qos level")
+	fs.StringVar(&o.MemLowQoSLevelConfigFile, "mem-low-qos-config-file",
+		o.MemLowQoSLevelConfigFile, "the absolute path of mem.low qos level config file")
 }
 
 func (o *MemoryOptions) ApplyTo(conf *qrmconfig.MemoryQRMPluginConfig) error {
@@ -98,5 +112,7 @@ func (o *MemoryOptions) ApplyTo(conf *qrmconfig.MemoryQRMPluginConfig) error {
 	conf.EnableSettingSockMem = o.EnableSettingSockMem
 	conf.SetGlobalTCPMemRatio = o.SetGlobalTCPMemRatio
 	conf.SetCgroupTCPMemRatio = o.SetCgroupTCPMemRatio
+	conf.EnableSettingMemLow = o.EnableSettingMemLow
+	conf.MemLowQoSLevelConfigFile = o.MemLowQoSLevelConfigFile
 	return nil
 }

pkg/agent/qrm-plugins/memory/dynamicpolicy/policy.go

@@ -40,6 +40,7 @@ import (
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/memory/dynamicpolicy/memoryadvisor"
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/memory/dynamicpolicy/oom"
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/memory/dynamicpolicy/state"
+	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/memory/handlers/memlow"
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/memory/handlers/sockmem"
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/util"
 	"github.com/kubewharf/katalyst-core/pkg/agent/utilcomponent/periodicalhandler"
@@ -146,6 +147,7 @@ type DynamicPolicy struct {
 
 	enableSettingMemoryMigrate bool
 	enableSettingSockMem       bool
+	enableSettingMemLow        bool
 	enableMemoryAdvisor        bool
 	memoryAdvisorSocketAbsPath string
 	memoryPluginSocketAbsPath  string
@@ -211,6 +213,7 @@ func NewDynamicPolicy(agentCtx *agent.GenericContext, conf *config.Configuration
 		defaultAsyncLimitedWorkers: asyncworker.NewAsyncLimitedWorkers(memoryPluginAsyncWorkersName, defaultAsyncWorkLimit, wrappedEmitter),
 		enableSettingMemoryMigrate: conf.EnableSettingMemoryMigrate,
 		enableSettingSockMem:       conf.EnableSettingSockMem,
+		enableSettingMemLow:        conf.EnableSettingMemLow,
 		enableMemoryAdvisor:        conf.EnableMemoryAdvisor,
 		memoryAdvisorSocketAbsPath: conf.MemoryAdvisorSocketAbsPath,
 		memoryPluginSocketAbsPath:  conf.MemoryPluginSocketAbsPath,
@@ -367,6 +370,15 @@ func (p *DynamicPolicy) Start() (err error) {
 		}
 	}
 
+	if p.enableSettingMemLow {
+		general.Infof("setMemLow enabled")
+		err := periodicalhandler.RegisterPeriodicalHandler(qrm.QRMMemoryPluginPeriodicalHandlerGroupName,
+			memlow.EnableSetMemLowPeriodicalHandlerName, memlow.MemLowTaskFunc, 90*time.Second)
+		if err != nil {
+			general.Infof("setMemLow failed, err=%v", err)
+		}
+	}
+
 	go wait.Until(func() {
 		periodicalhandler.ReadyToStartHandlersByGroup(qrm.QRMMemoryPluginPeriodicalHandlerGroupName)
 	}, 5*time.Second, p.stopCh)

pkg/agent/qrm-plugins/memory/dynamicpolicy/policy_test.go

@@ -902,6 +902,7 @@ func TestAllocate(t *testing.T) {
 		}
 
 		dynamicPolicy.enableMemoryAdvisor = true
+		dynamicPolicy.enableSettingMemLow = true
 		dynamicPolicy.advisorClient = advisorsvc.NewStubAdvisorServiceClient()
 
 		resp, err := dynamicPolicy.Allocate(context.Background(), tc.req)

pkg/agent/qrm-plugins/memory/handlers/memlow/const.go

@@ -0,0 +1,31 @@
+/*
+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 memlow
+
+const EnableSetMemLowPeriodicalHandlerName = "SetCGMemLow"
+
+const (
+	// Constants for cgroup memory statistics
+	cgroupMemory32M       = 33554432
+	cgroupMemoryUnlimited = 9223372036854771712
+
+	controlKnobKeyMemLow = "mem_low"
+)
+
+const (
+	metricNameMemLow = "async_handler_cgroup_memlow"
+)

pkg/agent/qrm-plugins/memory/handlers/memlow/memlow_linux.go

@@ -0,0 +1,200 @@
+//go:build linux
+// +build linux
+
+/*
+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 memlow
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"strconv"
+
+	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/commonstate"
+	coreconfig "github.com/kubewharf/katalyst-core/pkg/config"
+	dynamicconfig "github.com/kubewharf/katalyst-core/pkg/config/agent/dynamic"
+	"github.com/kubewharf/katalyst-core/pkg/metaserver"
+	"github.com/kubewharf/katalyst-core/pkg/metrics"
+	"github.com/kubewharf/katalyst-core/pkg/util/cgroup/common"
+	cgroupcm "github.com/kubewharf/katalyst-core/pkg/util/cgroup/common"
+	cgroupmgr "github.com/kubewharf/katalyst-core/pkg/util/cgroup/manager"
+	"github.com/kubewharf/katalyst-core/pkg/util/general"
+	"github.com/kubewharf/katalyst-core/pkg/util/native"
+)
+
+func calculatedBestLow(memUsage, memFileInactive, userLow uint64) uint64 {
+	if memFileInactive > memUsage {
+		return 0
+	}
+	minLow := memUsage - memFileInactive + cgroupMemory32M
+	low := math.Max(float64(userLow), float64(minLow))
+	return uint64(low)
+}
+
+func getUserSpecifiedMemoryLowInBytes(memLimit, memUsage, ratio uint64) uint64 {
+	if ratio > 100 || ratio <= 0 {
+		general.Infof("Bad ratio %v", ratio)
+		return 0
+	}
+
+	maxLimit := memLimit
+	if memLimit >= cgroupMemoryUnlimited {
+		maxLimit = memUsage + cgroupMemory32M
+	}
+	low := uint64(float64(maxLimit) / 100.0 * float64(ratio))
+	low = general.AlignToPageSize(low)
+	return low
+}
+
+func calculateMemLow(absCgPath string, ratio uint64) (uint64, error) {
+	/*
+	 * I hope to protect cgroup from System-Thrashing(insufficient hot file memory)
+	 * during kswapd reclaiming through cgv2 memory.low.
+	 */
+	// Step1, get cgroup memory.limit, memory.usage, inactive-file-memory.
+	memStat, err := cgroupmgr.GetMemoryWithAbsolutePath(absCgPath)
+	if err != nil {
+		general.Warningf("GetMemoryWithAbsolutePath failed with err: %v", err)
+		return 0, err
+	}
+	memDetailedStat, err := cgroupmgr.GetDetailedMemoryWithAbsolutePath(absCgPath)
+	if err != nil {
+		general.Warningf("GetDetailedMemoryWithAbsolutePath failed with err: %v", err)
+		return 0, err
+	}
+
+	// Step2, Reserve a certain ratio of file memory for high-QoS cgroups.
+	userLow := getUserSpecifiedMemoryLowInBytes(memStat.Limit, memStat.Usage, ratio)
+	if userLow == 0 {
+		general.Warningf("getUserSpecifiedMemoryLowInBytes return 0")
+		return 0, fmt.Errorf("getUserSpecifiedMemoryLowInBytes return 0")
+	}
+
+	// Step3, I don't want to hurt existing hot file-memory.
+	// If the reserve file memory is not sufficient for current hot file-memory,
+	// then the final memory.low will be based on current hot file-memory.
+	low := calculatedBestLow(memStat.Usage, memDetailedStat.FileInactive, userLow)
+
+	general.Infof("calculateMemLow: target=%v, usr=%v, ratio=%v, cg=%v, limit=%v, usage=%v, file-inactive=%v", low, userLow, ratio, absCgPath, memStat.Limit, memStat.Usage, memDetailedStat.FileInactive)
+	return low, nil
+}
+
+func applyMemLowQoSLevelConfig(conf *coreconfig.Configuration,
+	emitter metrics.MetricEmitter, metaServer *metaserver.MetaServer,
+) {
+	if conf.MemLowQoSLevelConfigFile == "" {
+		general.Infof("no MemLowQoSLevelConfigFile found")
+		return
+	}
+
+	var extraControlKnobConfigs commonstate.ExtraControlKnobConfigs
+	if err := general.LoadJsonConfig(conf.MemLowQoSLevelConfigFile, &extraControlKnobConfigs); err != nil {
+		general.Errorf("MemLowQoSLevelConfigFile load failed:%v", err)
+		return
+	}
+	ctx := context.Background()
+	podList, err := metaServer.GetPodList(ctx, native.PodIsActive)
+	if err != nil {
+		general.Infof("get pod list failed: %v", err)
+		return
+	}
+
+	for _, pod := range podList {
+		if pod == nil {
+			general.Warningf("get nil pod from metaServer")
+			continue
+		}
+		if conf.QoSConfiguration == nil {
+			continue
+		}
+		qosConfig := conf.QoSConfiguration
+		qosLevel, err := qosConfig.GetQoSLevelForPod(pod)
+		if err != nil {
+			general.Warningf("GetQoSLevelForPod failed:%v", err)
+			continue
+		}
+		qosLevelDefaultValue, ok := extraControlKnobConfigs[controlKnobKeyMemLow].QoSLevelToDefaultValue[qosLevel]
+		if !ok {
+			continue
+		}
+
+		ratio, err := strconv.ParseFloat(qosLevelDefaultValue, 64)
+		if err != nil {
+			general.Infof("Atoi failed with err: %v", err)
+			continue
+		}
+
+		absCgPath, err := common.GetPodAbsCgroupPath(common.CgroupSubsysMemory, string(pod.UID))
+		if err != nil {
+			continue
+		}
+		low, err := calculateMemLow(absCgPath, uint64(ratio*100))
+		if err != nil {
+			general.Errorf("calculateMemLow for relativeCgPath: %s failed with error: %v",
+				absCgPath, err)
+			continue
+		}
+
+		// OK. Set the value for memory.low.
+		var data *cgroupcm.MemoryData
+		data = &cgroupcm.MemoryData{SoftLimitInBytes: int64(low)}
+		if err := cgroupmgr.ApplyMemoryWithAbsolutePath(absCgPath, data); err != nil {
+			general.Warningf("ApplyMemoryWithRelativePath failed, cgpath=%v, err=%v", absCgPath, err)
+			continue
+		}
+		_ = emitter.StoreInt64(metricNameMemLow, int64(low), metrics.MetricTypeNameRaw,
+			metrics.ConvertMapToTags(map[string]string{
+				"podUID": string(pod.UID),
+			})...)
+
+	}
+}
+
+func MemLowTaskFunc(conf *coreconfig.Configuration,
+	_ interface{}, _ *dynamicconfig.DynamicAgentConfiguration,
+	emitter metrics.MetricEmitter, metaServer *metaserver.MetaServer,
+) {
+	general.Infof("called")
+
+	if conf == nil {
+		general.Errorf("nil extraConf")
+		return
+	} else if emitter == nil {
+		general.Errorf("nil emitter")
+		return
+	} else if metaServer == nil {
+		general.Errorf("nil metaServer")
+		return
+	}
+
+	// SettingMemLow featuregate.
+	if !conf.EnableSettingMemLow {
+		general.Infof("EnableSettingMemLow disabled")
+		return
+	}
+
+	if !cgroupcm.CheckCgroup2UnifiedMode() {
+		general.Infof("not in cgv2 environment, skip MemLowTaskFunc")
+		return
+	}
+
+	// checking qos-level memory.low configuration.
+	if len(conf.MemLowQoSLevelConfigFile) > 0 {
+		applyMemLowQoSLevelConfig(conf, emitter, metaServer)
+	}
+}