[WIP]feat(*): support user memory watermark

cmd/katalyst-agent/app/options/dynamic/dynamic_base.go

@@ -24,6 +24,7 @@ import (
 	"github.com/kubewharf/katalyst-core/cmd/katalyst-agent/app/options/dynamic/irqtuning"
 	"github.com/kubewharf/katalyst-core/cmd/katalyst-agent/app/options/dynamic/strategygroup"
 	"github.com/kubewharf/katalyst-core/cmd/katalyst-agent/app/options/dynamic/tmo"
+	"github.com/kubewharf/katalyst-core/cmd/katalyst-agent/app/options/dynamic/userwatermark"
 	"github.com/kubewharf/katalyst-core/pkg/config/agent/dynamic"
 )
 
@@ -32,6 +33,7 @@ type DynamicOptions struct {
 	*tmo.TransparentMemoryOffloadingOptions
 	*strategygroup.StrategyGroupOptions
 	*irqtuning.IRQTuningOptions
+	*userwatermark.UserWatermarkOptions
 }
 
 func NewDynamicOptions() *DynamicOptions {
@@ -40,6 +42,7 @@ func NewDynamicOptions() *DynamicOptions {
 		TransparentMemoryOffloadingOptions: tmo.NewTransparentMemoryOffloadingOptions(),
 		StrategyGroupOptions:               strategygroup.NewStrategyGroupOptions(),
 		IRQTuningOptions:                   irqtuning.NewIRQTuningOptions(),
+		UserWatermarkOptions:               userwatermark.NewUserWatermarkOptions(),
 	}
 }
 
@@ -48,6 +51,7 @@ func (o *DynamicOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 	o.TransparentMemoryOffloadingOptions.AddFlags(fss)
 	o.StrategyGroupOptions.AddFlags(fss)
 	o.IRQTuningOptions.AddFlags(fss)
+	o.UserWatermarkOptions.AddFlags(fss)
 }
 
 func (o *DynamicOptions) ApplyTo(c *dynamic.Configuration) error {
@@ -56,5 +60,6 @@ func (o *DynamicOptions) ApplyTo(c *dynamic.Configuration) error {
 	errList = append(errList, o.TransparentMemoryOffloadingOptions.ApplyTo(c.TransparentMemoryOffloadingConfiguration))
 	errList = append(errList, o.StrategyGroupOptions.ApplyTo(c.StrategyGroupConfiguration))
 	errList = append(errList, o.IRQTuningOptions.ApplyTo(c.IRQTuningConfiguration))
+	errList = append(errList, o.UserWatermarkOptions.ApplyTo(c.UserWatermarkConfiguration))
 	return errors.NewAggregate(errList)
 }

cmd/katalyst-agent/app/options/dynamic/userwatermark/user_watermark_base.go

@@ -0,0 +1,58 @@
+/*
+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 userwatermark
+
+import (
+	cliflag "k8s.io/component-base/cli/flag"
+
+	defaultOptions "github.com/kubewharf/katalyst-core/cmd/katalyst-agent/app/options/dynamic/userwatermark/userwatermarkdefault"
+	"github.com/kubewharf/katalyst-core/pkg/config/agent/dynamic/userwatermark"
+)
+
+type UserWatermarkOptions struct {
+	EnableReclaimer   bool
+	ReconcileInterval int64
+	ServiceLabel      string
+	*defaultOptions.DefaultOptions
+}
+
+func NewUserWatermarkOptions() *UserWatermarkOptions {
+	return &UserWatermarkOptions{
+		DefaultOptions: defaultOptions.NewDefaultOptions(),
+	}
+}
+
+func (o *UserWatermarkOptions) AddFlags(fss *cliflag.NamedFlagSets) {
+	fs := fss.FlagSet("user-watermark")
+
+	fs.BoolVar(&o.EnableReclaimer, "enable-user-watermark-reclaimer", o.EnableReclaimer,
+		"whether to enable memory reclaimer")
+	fs.Int64Var(&o.ReconcileInterval, "user-watermark-reconcile-interval", o.ReconcileInterval,
+		"the interval to reconcile memory")
+	fs.StringVar(&o.ServiceLabel, "user-watermark-pod-service-label", o.ServiceLabel,
+		"the service label to filter")
+
+	o.DefaultOptions.AddFlags(fss)
+}
+
+func (o *UserWatermarkOptions) ApplyTo(c *userwatermark.UserWatermarkConfiguration) error {
+	c.EnableReclaimer = o.EnableReclaimer
+	c.ReconcileInterval = o.ReconcileInterval
+	c.ServiceLabel = o.ServiceLabel
+
+	return o.DefaultOptions.ApplyTo(c.DefaultConfig)
+}

cmd/katalyst-agent/app/options/dynamic/userwatermark/user_watermark_base_test.go

@@ -0,0 +1,67 @@
+/*
+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 userwatermark
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	cliflag "k8s.io/component-base/cli/flag"
+
+	dynamicuserwm "github.com/kubewharf/katalyst-core/pkg/config/agent/dynamic/userwatermark"
+)
+
+func TestNewUserWatermarkOptions(t *testing.T) {
+	t.Parallel()
+
+	opts := NewUserWatermarkOptions()
+	assert.NotNil(t, opts)
+	assert.NotNil(t, opts.DefaultOptions)
+}
+
+func TestUserWatermarkOptions_AddFlags(t *testing.T) {
+	t.Parallel()
+
+	opts := NewUserWatermarkOptions()
+	fss := &cliflag.NamedFlagSets{}
+
+	opts.AddFlags(fss)
+	fs := fss.FlagSet("user-watermark")
+	assert.NotNil(t, fs.Lookup("enable-user-watermark-reclaimer"))
+	assert.NotNil(t, fs.Lookup("user-watermark-reconcile-interval"))
+	assert.NotNil(t, fs.Lookup("user-watermark-pod-service-label"))
+}
+
+func TestUserWatermarkOptions_ApplyTo(t *testing.T) {
+	t.Parallel()
+
+	opts := &UserWatermarkOptions{
+		EnableReclaimer:   true,
+		ReconcileInterval: 10,
+		ServiceLabel:      "service-label",
+		DefaultOptions:    NewUserWatermarkOptions().DefaultOptions,
+	}
+
+	conf := dynamicuserwm.NewUserWatermarkConfiguration()
+	err := opts.ApplyTo(conf)
+
+	assert.NoError(t, err)
+	assert.True(t, conf.EnableReclaimer)
+	assert.Equal(t, int64(10), conf.ReconcileInterval)
+	assert.Equal(t, "service-label", conf.ServiceLabel)
+	assert.NotNil(t, conf.DefaultConfig)
+}

cmd/katalyst-agent/app/options/dynamic/userwatermark/userwatermarkdefault/user_watermark_default_config.go

@@ -0,0 +1,94 @@
+/*
+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 userwatermarkdefault
+
+import (
+	"time"
+
+	cliflag "k8s.io/component-base/cli/flag"
+
+	"github.com/kubewharf/katalyst-api/pkg/apis/config/v1alpha1"
+	"github.com/kubewharf/katalyst-core/pkg/config/agent/dynamic/userwatermark"
+)
+
+type DefaultOptions struct {
+	EnableMemoryReclaim bool
+	ReclaimInterval     int64
+
+	ScaleFactor         uint64
+	SingleReclaimFactor float64
+	// SingleReclaimSize is the max memory reclaim size in one reclaim cycle
+	SingleReclaimSize      uint64
+	BackoffDuration        time.Duration
+	FeedbackPolicy         string
+	ReclaimFailedThreshold uint64
+	FailureFreezePeriod    time.Duration
+
+	PsiAvg60Threshold           float64
+	ReclaimAccuracyTarget       float64
+	ReclaimScanEfficiencyTarget float64
+}
+
+func NewDefaultOptions() *DefaultOptions {
+	return &DefaultOptions{}
+}
+
+func (o *DefaultOptions) AddFlags(fss *cliflag.NamedFlagSets) {
+	fs := fss.FlagSet("user-watermark")
+
+	fs.BoolVar(&o.EnableMemoryReclaim, "enable-user-watermark-memory-reclaim", o.EnableMemoryReclaim,
+		"whether to enable memory reclaim")
+	fs.Int64Var(&o.ReclaimInterval, "user-watermark-reclaim-interval", o.ReclaimInterval,
+		"the interval to reclaim memory")
+	fs.Uint64Var(&o.ScaleFactor, "user-watermark-scale-factor", o.ScaleFactor,
+		"the scale factor to reclaim memory")
+	fs.Uint64Var(&o.SingleReclaimSize, "user-watermark-single-reclaim-size", o.SingleReclaimSize,
+		"the max memory reclaim size in one reclaim cycle")
+	fs.Float64Var(&o.SingleReclaimFactor, "user-watermark-single-reclaim-factor", o.SingleReclaimFactor,
+		"the factor to reclaim memory")
+	fs.DurationVar(&o.BackoffDuration, "user-watermark-backoff-duration", o.BackoffDuration,
+		"the duration to backoff after reclaim failed")
+	fs.StringVar(&o.FeedbackPolicy, "user-watermark-feedback-policy", o.FeedbackPolicy,
+		"the feedback policy to reclaim memory")
+	fs.Uint64Var(&o.ReclaimFailedThreshold, "user-watermark-reclaim-failed-threshold", o.ReclaimFailedThreshold,
+		"the threshold to trigger reclaim failed")
+	fs.DurationVar(&o.FailureFreezePeriod, "user-watermark-failure-freeze-period", o.FailureFreezePeriod,
+		"the period to freeze reclaim after trigger reclaim failed")
+	fs.Float64Var(&o.PsiAvg60Threshold, "user-watermark-psi-avg60-threshold", o.PsiAvg60Threshold,
+		"the threshold to trigger reclaim failed")
+	fs.Float64Var(&o.ReclaimAccuracyTarget, "user-watermark-reclaim-accuracy-target", o.ReclaimAccuracyTarget,
+		"the target reclaim accuracy")
+	fs.Float64Var(&o.ReclaimScanEfficiencyTarget, "user-watermark-reclaim-scan-efficiency-target", o.ReclaimScanEfficiencyTarget,
+		"the target reclaim scan efficiency")
+}
+
+func (o *DefaultOptions) ApplyTo(c *userwatermark.UserWatermarkDefaultConfiguration) error {
+	c.EnableMemoryReclaim = o.EnableMemoryReclaim
+	c.ReclaimInterval = o.ReclaimInterval
+
+	c.ScaleFactor = o.ScaleFactor
+	c.SingleReclaimSize = o.SingleReclaimSize
+	c.SingleReclaimFactor = o.SingleReclaimFactor
+	c.BackoffDuration = o.BackoffDuration
+	c.FeedbackPolicy = v1alpha1.UserWatermarkPolicyName(o.FeedbackPolicy)
+	c.ReclaimFailedThreshold = o.ReclaimFailedThreshold
+	c.FailureFreezePeriod = o.FailureFreezePeriod
+	c.PsiAvg60Threshold = o.PsiAvg60Threshold
+	c.ReclaimAccuracyTarget = o.ReclaimAccuracyTarget
+	c.ReclaimScanEfficiencyTarget = o.ReclaimScanEfficiencyTarget
+	return nil
+}

cmd/katalyst-agent/app/options/dynamic/userwatermark/userwatermarkdefault/user_watermark_default_config_test.go

@@ -0,0 +1,96 @@
+/*
+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 userwatermarkdefault
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	cliflag "k8s.io/component-base/cli/flag"
+
+	v1alpha1 "github.com/kubewharf/katalyst-api/pkg/apis/config/v1alpha1"
+	dynamicuserwm "github.com/kubewharf/katalyst-core/pkg/config/agent/dynamic/userwatermark"
+)
+
+func TestNewDefaultOptions(t *testing.T) {
+	t.Parallel()
+
+	opts := NewDefaultOptions()
+	assert.NotNil(t, opts)
+}
+
+func TestDefaultOptions_AddFlags(t *testing.T) {
+	t.Parallel()
+
+	opts := NewDefaultOptions()
+	fss := &cliflag.NamedFlagSets{}
+
+	opts.AddFlags(fss)
+
+	fs := fss.FlagSet("user-watermark")
+
+	assert.NotNil(t, fs.Lookup("enable-user-watermark-memory-reclaim"))
+	assert.NotNil(t, fs.Lookup("user-watermark-reclaim-interval"))
+	assert.NotNil(t, fs.Lookup("user-watermark-scale-factor"))
+	assert.NotNil(t, fs.Lookup("user-watermark-single-reclaim-size"))
+	assert.NotNil(t, fs.Lookup("user-watermark-single-reclaim-factor"))
+	assert.NotNil(t, fs.Lookup("user-watermark-backoff-duration"))
+
+	assert.NotNil(t, fs.Lookup("user-watermark-feedback-policy"))
+	assert.NotNil(t, fs.Lookup("user-watermark-reclaim-failed-threshold"))
+	assert.NotNil(t, fs.Lookup("user-watermark-failure-freeze-period"))
+	assert.NotNil(t, fs.Lookup("user-watermark-psi-avg60-threshold"))
+	assert.NotNil(t, fs.Lookup("user-watermark-reclaim-accuracy-target"))
+	assert.NotNil(t, fs.Lookup("user-watermark-reclaim-scan-efficiency-target"))
+}
+
+func TestDefaultOptions_ApplyTo(t *testing.T) {
+	t.Parallel()
+
+	opts := &DefaultOptions{
+		EnableMemoryReclaim: true,
+		ReclaimInterval:     5,
+		ScaleFactor:         200,
+		SingleReclaimFactor: 0.5,
+		SingleReclaimSize:   1 << 20,
+		BackoffDuration:     10,
+		FeedbackPolicy:      string(v1alpha1.UserWatermarkPolicyNamePSI),
+
+		ReclaimFailedThreshold:      3,
+		FailureFreezePeriod:         20,
+		PsiAvg60Threshold:           1.0,
+		ReclaimAccuracyTarget:       0.8,
+		ReclaimScanEfficiencyTarget: 0.5,
+	}
+
+	conf := &dynamicuserwm.UserWatermarkDefaultConfiguration{}
+	err := opts.ApplyTo(conf)
+
+	assert.NoError(t, err)
+	assert.True(t, conf.EnableMemoryReclaim)
+	assert.Equal(t, int64(5), conf.ReclaimInterval)
+	assert.Equal(t, uint64(200), conf.ScaleFactor)
+	assert.Equal(t, 0.5, conf.SingleReclaimFactor)
+	assert.Equal(t, uint64(1<<20), conf.SingleReclaimSize)
+	assert.Equal(t, opts.BackoffDuration, conf.BackoffDuration)
+	assert.Equal(t, v1alpha1.UserWatermarkPolicyNamePSI, conf.FeedbackPolicy)
+	assert.Equal(t, uint64(3), conf.ReclaimFailedThreshold)
+	assert.Equal(t, opts.FailureFreezePeriod, conf.FailureFreezePeriod)
+	assert.Equal(t, opts.PsiAvg60Threshold, conf.PsiAvg60Threshold)
+	assert.Equal(t, opts.ReclaimAccuracyTarget, conf.ReclaimAccuracyTarget)
+	assert.Equal(t, opts.ReclaimScanEfficiencyTarget, conf.ReclaimScanEfficiencyTarget)
+}

go.mod

@@ -175,6 +175,7 @@ require (
 )
 
 replace (
+	github.com/kubewharf/katalyst-api => github.com/JulyWindK/katalyst-api v1.6.9
 	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

@@ -62,6 +62,8 @@ github.com/Djarvur/go-err113 v0.0.0-20200511133814-5174e21577d5/go.mod h1:4UJr5H
 github.com/GoogleCloudPlatform/k8s-cloud-provider v1.16.1-0.20210702024009-ea6160c1d0e3/go.mod h1:8XasY4ymP2V/tn2OOV9ZadmiTE1FIB/h3W+yNlPttKw=
 github.com/HdrHistogram/hdrhistogram-go v1.0.0/go.mod h1:YzE1EgsuAz8q9lfGdlxBZo2Ma655+PfKp2mlzcAqIFw=
 github.com/JeffAshton/win_pdh v0.0.0-20161109143554-76bb4ee9f0ab/go.mod h1:3VYc5hodBMJ5+l/7J4xAyMeuM2PNuepvHlGs8yilUCA=
+github.com/JulyWindK/katalyst-api v1.6.9 h1:NRQOSQmeg6A+EuNpc8WgzMYceFKeqVlv4OnKHvhmKZY=
+github.com/JulyWindK/katalyst-api v1.6.9/go.mod h1:BZMVGVl3EP0eCn5xsDgV41/gjYkoh43abIYxrB10e3k=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
 github.com/MakeNowJust/heredoc v0.0.0-20170808103936-bb23615498cd/go.mod h1:64YHyfSL2R96J44Nlwm39UHepQbyR5q10x7iYa1ks2E=
 github.com/Masterminds/semver v1.5.0/go.mod h1:MB6lktGJrhw8PrUyiEoblNEGEQ+RzHPF078ddwwvV3Y=
@@ -574,8 +576,6 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/kubewharf/katalyst-api v0.5.9-0.20260108125536-85e136f5902c h1:ohKHA5TOlW9487menKnKH2M14LeIq1xQ1yW4xp8x9o8=
-github.com/kubewharf/katalyst-api v0.5.9-0.20260108125536-85e136f5902c/go.mod h1:BZMVGVl3EP0eCn5xsDgV41/gjYkoh43abIYxrB10e3k=
 github.com/kubewharf/kubelet v1.24.6-kubewharf-pre.1 h1:pzU37yZWrOBosNX+Laay9Ess0Bff/rsWanBxbdXnHnM=
 github.com/kubewharf/kubelet v1.24.6-kubewharf-pre.1/go.mod h1:MxbSZUx3wXztFneeelwWWlX7NAAStJ6expqq7gY2J3c=
 github.com/kyoh86/exportloopref v0.1.7/go.mod h1:h1rDl2Kdj97+Kwh4gdz3ujE7XHmH51Q0lUiZ1z4NLj8=