Skip to content

Commit

Permalink
feat: adding support to qos level memory.low protection
Browse files Browse the repository at this point in the history
This patch provied the feature of memory.low protection.
There are a couple of benefits about memory.low protection:

1, it provides a gradient of protection.
 As a cgroup's usage grows past the protected amount,
 the protected amount remains protected,
 but reclaim pressure for the excess amount gradually increases.
2, it's work-conserving - if the protected cgroup doesn't use the memory,
 it's available for others to use.

Signed-off-by: Robin Lu <[email protected]>
  • Loading branch information
lubinszARM committed Jul 25, 2024
1 parent 20d207f commit 5117199
Show file tree
Hide file tree
Showing 9 changed files with 704 additions and 0 deletions.
16 changes: 16 additions & 0 deletions cmd/katalyst-agent/app/options/qrm/memory_plugin.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ type MemoryOptions struct {
OOMPriorityPinnedMapAbsPath string

SockMemOptions
MemProtectionOptions
}

type SockMemOptions struct {
Expand All @@ -43,6 +44,11 @@ type SockMemOptions struct {
SetCgroupTCPMemRatio int
}

type MemProtectionOptions struct {
EnableSettingMemProtection bool
MemProtectionQoSLevelConfigFile string
}

func NewMemoryOptions() *MemoryOptions {
return &MemoryOptions{
PolicyName: "dynamic",
Expand All @@ -56,6 +62,10 @@ func NewMemoryOptions() *MemoryOptions {
SetGlobalTCPMemRatio: 20, // default: 20% * {host total memory}
SetCgroupTCPMemRatio: 100, // default: 100% * {cgroup memory}
},
MemProtectionOptions: MemProtectionOptions{
EnableSettingMemProtection: false,
MemProtectionQoSLevelConfigFile: "",
},
}
}

Expand Down Expand Up @@ -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.EnableSettingMemProtection, "enable-setting-mem-protection",
o.EnableSettingMemProtection, "if set true, we will do memory protection in qos level")
fs.StringVar(&o.MemProtectionQoSLevelConfigFile, "mem-protection-qos-config-file",
o.MemProtectionQoSLevelConfigFile, "the absolute path of mem.low qos level config file")
}

func (o *MemoryOptions) ApplyTo(conf *qrmconfig.MemoryQRMPluginConfig) error {
Expand All @@ -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.EnableSettingMemProtection = o.EnableSettingMemProtection
conf.MemProtectionQoSLevelConfigFile = o.MemProtectionQoSLevelConfigFile
return nil
}
12 changes: 12 additions & 0 deletions pkg/agent/qrm-plugins/memory/dynamicpolicy/policy.go
Original file line number Diff line number Diff line change
Expand Up @@ -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/memprotection"
"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"
Expand Down Expand Up @@ -144,6 +145,7 @@ type DynamicPolicy struct {

enableSettingMemoryMigrate bool
enableSettingSockMem bool
enableSettingMemProtection bool
enableMemoryAdvisor bool
memoryAdvisorSocketAbsPath string
memoryPluginSocketAbsPath string
Expand Down Expand Up @@ -207,6 +209,7 @@ func NewDynamicPolicy(agentCtx *agent.GenericContext, conf *config.Configuration
defaultAsyncLimitedWorkers: asyncworker.NewAsyncLimitedWorkers(memoryPluginAsyncWorkersName, defaultAsyncWorkLimit, wrappedEmitter),
enableSettingMemoryMigrate: conf.EnableSettingMemoryMigrate,
enableSettingSockMem: conf.EnableSettingSockMem,
enableSettingMemProtection: conf.EnableSettingMemProtection,
enableMemoryAdvisor: conf.EnableMemoryAdvisor,
memoryAdvisorSocketAbsPath: conf.MemoryAdvisorSocketAbsPath,
memoryPluginSocketAbsPath: conf.MemoryPluginSocketAbsPath,
Expand Down Expand Up @@ -363,6 +366,15 @@ func (p *DynamicPolicy) Start() (err error) {
}
}

if p.enableSettingMemProtection {
general.Infof("setMemProtection enabled")
err := periodicalhandler.RegisterPeriodicalHandler(qrm.QRMMemoryPluginPeriodicalHandlerGroupName,
memprotection.EnableSetMemProtectionPeriodicalHandlerName, memprotection.MemProtectionTaskFunc, 90*time.Second)
if err != nil {
general.Infof("setMemProtection failed, err=%v", err)
}
}

go wait.Until(func() {
periodicalhandler.ReadyToStartHandlersByGroup(qrm.QRMMemoryPluginPeriodicalHandlerGroupName)
}, 5*time.Second, p.stopCh)
Expand Down
1 change: 1 addition & 0 deletions pkg/agent/qrm-plugins/memory/dynamicpolicy/policy_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -893,6 +893,7 @@ func TestAllocate(t *testing.T) {
}

dynamicPolicy.enableMemoryAdvisor = true
dynamicPolicy.enableSettingMemProtection = true
dynamicPolicy.advisorClient = advisorsvc.NewStubAdvisorServiceClient()

resp, err := dynamicPolicy.Allocate(context.Background(), tc.req)
Expand Down
31 changes: 31 additions & 0 deletions pkg/agent/qrm-plugins/memory/handlers/memprotection/const.go
Original file line number Diff line number Diff line change
@@ -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 memprotection

const EnableSetMemProtectionPeriodicalHandlerName = "SetCGMemProtection"

const (
// Constants for cgroup memory statistics
cgroupMemory32M = 33554432
cgroupMemoryUnlimited = 9223372036854771712

controlKnobKeyMemProtection = "mem_protection"
)

const (
metricNameMemLow = "async_handler_cgroup_memlow"
)
Original file line number Diff line number Diff line change
@@ -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 memprotection

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 calculatedBestProtection(memUsage, memFileInactive, userProtection uint64) uint64 {
if memFileInactive > memUsage {
return 0
}
minProtection := memUsage - memFileInactive + cgroupMemory32M
low := math.Max(float64(userProtection), float64(minProtection))
return uint64(low)
}

func getUserSpecifiedMemoryProtectionInBytes(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 calculateMemProtection(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.
userProtection := getUserSpecifiedMemoryProtectionInBytes(memStat.Limit, memStat.Usage, ratio)
if userProtection == 0 {
general.Warningf("getUserSpecifiedMemoryProtectionBytes return 0")
return 0, fmt.Errorf("getUserSpecifiedMemoryProtectionBytes 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 := calculatedBestProtection(memStat.Usage, memDetailedStat.FileInactive, userProtection)

general.Infof("calculateMemProtection: target=%v, usr=%v, ratio=%v, cg=%v, limit=%v, usage=%v, file-inactive=%v", low, userProtection, ratio, absCgPath, memStat.Limit, memStat.Usage, memDetailedStat.FileInactive)
return low, nil
}

func applyMemProtectionQoSLevelConfig(conf *coreconfig.Configuration,
emitter metrics.MetricEmitter, metaServer *metaserver.MetaServer,
) {
if conf.MemProtectionQoSLevelConfigFile == "" {
general.Infof("no MemProtectionQoSLevelConfigFile found")
return
}

var extraControlKnobConfigs commonstate.ExtraControlKnobConfigs
if err := general.LoadJsonConfig(conf.MemProtectionQoSLevelConfigFile, &extraControlKnobConfigs); err != nil {
general.Errorf("MemProtectionQoSLevelConfigFile 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[controlKnobKeyMemProtection].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 := calculateMemProtection(absCgPath, uint64(ratio*100))
if err != nil {
general.Errorf("calculateMemProtection 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.ApplyMemoryWithRelativePath(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 MemProtectionTaskFunc(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
}

// SettingMemProtection featuregate.
if !conf.EnableSettingMemProtection {
general.Infof("EnableSettingMemProtection disabled")
return
}

if !cgroupcm.CheckCgroup2UnifiedMode() {
general.Infof("not in cgv2 environment, skip MemProtectionTaskFunc")
return
}

// checking qos-level memory.low configuration.
if len(conf.MemProtectionQoSLevelConfigFile) > 0 {
applyMemProtectionQoSLevelConfig(conf, emitter, metaServer)
}
}
Loading

0 comments on commit 5117199

Please sign in to comment.