fix(qrm): fix

Signed-off-by: ricky <[email protected]>

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

@@ -23,12 +23,14 @@ import (
 )
 
 type MemoryOptions struct {
-	PolicyName                 string
-	ReservedMemoryGB           uint64
-	SkipMemoryStateCorruption  bool
-	EnableSettingMemoryMigrate bool
-	EnableMemoryAdvisor        bool
-	ExtraControlKnobConfigFile string
+	PolicyName                  string
+	ReservedMemoryGB            uint64
+	SkipMemoryStateCorruption   bool
+	EnableSettingMemoryMigrate  bool
+	EnableMemoryAdvisor         bool
+	ExtraControlKnobConfigFile  string
+	EnableOOMPriority           bool
+	OOMPriorityPinnedMapAbsPath string
 
 	SockMemOptions
 }
@@ -39,14 +41,6 @@ type SockMemOptions struct {
 	SetGlobalTCPMemRatio int
 	// SetCgroupTCPMemLimitRatio limit cgroup max tcp memory usage.
 	SetCgroupTCPMemRatio int
-	PolicyName                  string
-	ReservedMemoryGB            uint64
-	SkipMemoryStateCorruption   bool
-	EnableSettingMemoryMigrate  bool
-	EnableMemoryAdvisor         bool
-	ExtraControlKnobConfigFile  string
-	EnableOOMPriority           bool
-	OOMPriorityPinnedMapAbsPath string
 }
 
 func NewMemoryOptions() *MemoryOptions {

go.mod

@@ -143,7 +143,6 @@ require (
 )
 
 replace (
-	github.com/kubewharf/katalyst-api => github.com/y-ykcir/katalyst-api v0.0.0-20231113041648-d84cc2a3b5fd
 	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

@@ -872,8 +872,6 @@ github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xlab/treeprint v0.0.0-20181112141820-a009c3971eca/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
-github.com/y-ykcir/katalyst-api v0.0.0-20231113041648-d84cc2a3b5fd h1:3w0FMIMqO5IuOSv8Gvtl4Ho+on8K90eCV6pZSrQTk9o=
-github.com/y-ykcir/katalyst-api v0.0.0-20231113041648-d84cc2a3b5fd/go.mod h1:iVILS5UL5PRtkUPH2Iu1K/gFGTPMNItnth5fmQ80VGE=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

pkg/agent/qrm-plugins/memory/dynamicpolicy/oom/calculator.go

@@ -18,6 +18,7 @@ package oom
 
 import (
 	"fmt"
+	"sync"
 
 	core "k8s.io/api/core/v1"
 
@@ -27,15 +28,23 @@ import (
 
 type OOMPriorityCalculator func(qosConfig *generic.QoSConfiguration, pod *core.Pod, previousPriority int) (int, error)
 
-var oomPriorityCalculators = []OOMPriorityCalculator{
-	nativeOOMPriorityCalculator,
-}
+var (
+	oomPriorityCalculatorsMtx sync.RWMutex
+	oomPriorityCalculators    = []OOMPriorityCalculator{
+		nativeOOMPriorityCalculator,
+	}
+)
 
 func AppendOOMPriorityCalculator(calculator OOMPriorityCalculator) {
+	oomPriorityCalculatorsMtx.Lock()
+	defer oomPriorityCalculatorsMtx.Unlock()
 	oomPriorityCalculators = append(oomPriorityCalculators, calculator)
 }
 
 func GetOOMPriority(qosConfig *generic.QoSConfiguration, pod *core.Pod) (oomPriority int, err error) {
+	oomPriorityCalculatorsMtx.RLock()
+	defer oomPriorityCalculatorsMtx.RUnlock()
+
 	for _, calculator := range oomPriorityCalculators {
 		oomPriority, err = calculator(qosConfig, pod, oomPriority)
 		if err != nil {
@@ -51,7 +60,10 @@ func nativeOOMPriorityCalculator(qosConfig *generic.QoSConfiguration, pod *core.
 		return 0, err
 	}
 
-	userSpecifiedOOMPriority, _ := qos.GetOOMPriority(qosConfig, pod)
+	userSpecifiedOOMPriority, invalid := qos.GetOOMPriority(qosConfig, pod)
+	if invalid {
+		return 0, fmt.Errorf("pod %+v/%+v set invalid oom priority", pod.Namespace, pod.Name)
+	}
 
 	oomPriority := qos.AlignOOMPriority(qosLevel, userSpecifiedOOMPriority)
 

pkg/agent/qrm-plugins/memory/dynamicpolicy/oom/calculator_test.go

@@ -45,6 +45,7 @@ func TestNativeOOMPriorityCalculator(t *testing.T) {
 		pod               *v1.Pod
 		expectAnnotations map[string]string
 		want              int
+		wantErr           bool
 	}{
 		{
 			name: "pod without both qos level annotation and oom priority annotation",
@@ -54,7 +55,8 @@ func TestNativeOOMPriorityCalculator(t *testing.T) {
 					Annotations: map[string]string{},
 				},
 			},
-			want: 0,
+			want:    0,
+			wantErr: false,
 		},
 		{
 			name: "pod with qos level annotation but without oom priority annotation",
@@ -66,7 +68,8 @@ func TestNativeOOMPriorityCalculator(t *testing.T) {
 					},
 				},
 			},
-			want: 100,
+			want:    100,
+			wantErr: false,
 		},
 		{
 			name: "pod with qos level annotation but with invalid oom priority annotation",
@@ -79,7 +82,8 @@ func TestNativeOOMPriorityCalculator(t *testing.T) {
 					},
 				},
 			},
-			want: 200,
+			want:    0,
+			wantErr: true,
 		},
 		{
 			name: "pod with qos level annotation but with oom priority annotation out of range",
@@ -92,7 +96,8 @@ func TestNativeOOMPriorityCalculator(t *testing.T) {
 					},
 				},
 			},
-			want: -1,
+			want:    -1,
+			wantErr: false,
 		},
 		{
 			name: "pod with qos level annotation and with correct oom priority annotation",
@@ -105,14 +110,15 @@ func TestNativeOOMPriorityCalculator(t *testing.T) {
 					},
 				},
 			},
-			want: -20,
+			want:    -20,
+			wantErr: false,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			got, err := nativeOOMPriorityCalculator(qosConfig, tt.pod, 0)
-			if err != nil {
+			if (err != nil) != tt.wantErr {
 				t.Errorf("nativeOOMPriorityCalculator() error = %v", err)
 				return
 			}

pkg/agent/qrm-plugins/memory/dynamicpolicy/oom/const.go

@@ -0,0 +1,22 @@
+/*
+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 oom
+
+const (
+	ClearResidualOOMPriorityPeriodicalHandlerName = "clearResidualOOMPriority"
+	SyncOOMPriorityPriorityPeriodicalHandlerName  = "syncOOMPriority"
+)

pkg/agent/qrm-plugins/memory/dynamicpolicy/oom/init.go

@@ -16,14 +16,27 @@ limitations under the License.
 
 package oom
 
+import "sync"
+
 type InitOOMPriorityBPFFunc func() error
 
-var initOOMPriorityBPF InitOOMPriorityBPFFunc
+var (
+	initOOMPriorityBPF    InitOOMPriorityBPFFunc
+	initOOMPriorityBPFMtx sync.RWMutex
+)
 
 func SetInitOOMPriorityBPFFunc(f InitOOMPriorityBPFFunc) {
+	initOOMPriorityBPFMtx.Lock()
+	defer initOOMPriorityBPFMtx.Unlock()
 	initOOMPriorityBPF = f
 }
 
 func GetInitOOMPriorityBPFFunc() InitOOMPriorityBPFFunc {
+	initOOMPriorityBPFMtx.RLock()
+	defer initOOMPriorityBPFMtx.RUnlock()
 	return initOOMPriorityBPF
 }
+
+func DummyInitOOMPriorityBPF() error {
+	return nil
+}

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

@@ -37,6 +37,7 @@ import (
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/advisorsvc"
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/commonstate"
 	"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/sockmem"
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/util"
@@ -211,8 +212,6 @@ func NewDynamicPolicy(agentCtx *agent.GenericContext, conf *config.Configuration
 	}
 
 	if policyImplement.enableOOMPriority {
-		go policyImplement.PollOOMBPFInit()
-
 		policyImplement.enhancementHandlers.Register(apiconsts.QRMPhaseRemovePod,
 			apiconsts.PodAnnotationMemoryEnhancementOOMPriority, policyImplement.clearOOMPriority)
 	}
@@ -269,8 +268,19 @@ func (p *DynamicPolicy) Start() (err error) {
 
 	if p.enableOOMPriority {
 		general.Infof("OOM priority enabled")
-		go wait.Until(p.clearResidualOOMPriority, clearOOMPriorityPeriod, p.stopCh)
-		go wait.Until(p.syncOOMPriority, syncOOMPriorityPeriod, p.stopCh)
+		go p.PollOOMBPFInit(p.stopCh)
+
+		err := periodicalhandler.RegisterPeriodicalHandler(qrm.QRMMemoryPluginPeriodicalHandlerGroupName,
+			oom.ClearResidualOOMPriorityPeriodicalHandlerName, p.clearResidualOOMPriority, clearOOMPriorityPeriod)
+		if err != nil {
+			general.Infof("register clearResidualOOMPriority failed, err=%v", err)
+		}
+
+		err = periodicalhandler.RegisterPeriodicalHandler(qrm.QRMMemoryPluginPeriodicalHandlerGroupName,
+			oom.SyncOOMPriorityPriorityPeriodicalHandlerName, p.syncOOMPriority, syncOOMPriorityPeriod)
+		if err != nil {
+			general.Infof("register syncOOMPriority failed, err=%v", err)
+		}
 	}
 
 	if p.enableSettingSockMem {