Merge pull request #870 from h-w-chen/dev/pap-cpufreq-as-dvfs-limiter

enhance(advisor): power_aware advisor - cpufreq as dvfs limiter

Author: xu282934741

cmd/katalyst-agent/app/options/metaserver/metric.go

@@ -55,8 +55,11 @@ func NewMetricFetcherOptions() *MetricFetcherOptions {
 		MetricInsurancePeriod: defaultMetricInsurancePeriod,
 		MetricProvisions:      []string{metaserver.MetricProvisionerMalachite, metaserver.MetricProvisionerKubelet},
 
-		DefaultInterval:         time.Second * 5,
-		ProvisionerIntervalSecs: map[string]int{metaserver.MetricProvisionerMalachiteRealtime: 1},
+		DefaultInterval: time.Second * 5,
+		ProvisionerIntervalSecs: map[string]int{
+			metaserver.MetricProvisionerMalachiteRealtime:     1,
+			metaserver.MetricProvisionerMalachiteRealtimeFreq: 1,
+		},
 
 		MalachiteOptions: &MalachiteOptions{},
 		CgroupOptions:    &CgroupOptions{},

cmd/katalyst-agent/app/options/sysadvisor/poweraware/power_aware_plugin.go

@@ -28,6 +28,7 @@ type PowerAwarePluginOptions struct {
 	DisablePowerPressureEvict        bool
 	PowerCappingAdvisorSocketAbsPath string
 	AnnotationKeyPrefix              string
+	DVFSIndication                   string
 }
 
 func (p *PowerAwarePluginOptions) AddFlags(fss *cliflag.NamedFlagSets) {
@@ -37,6 +38,7 @@ func (p *PowerAwarePluginOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 	fs.BoolVar(&p.DisablePowerCapping, "power-capping-Disabled", p.DisablePowerCapping, "flag for power aware plugin disabling power capping")
 	fs.StringVar(&p.PowerCappingAdvisorSocketAbsPath, "power-capping-advisor-sock-abs-path", p.PowerCappingAdvisorSocketAbsPath, "absolute path of unix socket file for power capping advisor served in sys-advisor")
 	fs.StringVar(&p.AnnotationKeyPrefix, "power-aware-annotation-key-prefix", p.AnnotationKeyPrefix, "prefix of node annotation keys used by power aware plugin")
+	fs.StringVar(&p.DVFSIndication, "power-aware-dvfs-indication", p.DVFSIndication, "indication metric name of dvfs effect")
 }
 
 func (p *PowerAwarePluginOptions) ApplyTo(o *poweraware.PowerAwarePluginConfiguration) error {
@@ -45,11 +47,14 @@ func (p *PowerAwarePluginOptions) ApplyTo(o *poweraware.PowerAwarePluginConfigur
 	o.DisablePowerCapping = p.DisablePowerCapping
 	o.PowerCappingAdvisorSocketAbsPath = p.PowerCappingAdvisorSocketAbsPath
 	o.AnnotationKeyPrefix = p.AnnotationKeyPrefix
+	o.DVFSIndication = p.DVFSIndication
 
 	return nil
 }
 
 // NewPowerAwarePluginOptions creates a new Options with a default config.
 func NewPowerAwarePluginOptions() *PowerAwarePluginOptions {
-	return &PowerAwarePluginOptions{}
+	return &PowerAwarePluginOptions{
+		DVFSIndication: poweraware.DVFSIndicationPower,
+	}
 }

pkg/agent/sysadvisor/plugin/poweraware/advisor/action/strategy/assess/cpufreq_change.go

@@ -0,0 +1,147 @@
+/*
+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 assess
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pkg/errors"
+
+	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/poweraware/reader"
+	"github.com/kubewharf/katalyst-core/pkg/util/general"
+)
+
+const (
+	// minKHZ is the minimum khz for a valid cpu freq
+	minKHZ = 1000_000
+
+	// alpha is the decaying coefficient of Exponential Moving Average formula
+	alpha = 0.4
+)
+
+type cpuFreqChangeAssessor struct {
+	// highFreqKHZ may update itself to higher value observed
+	// todo: consider to get fixed base cpu freq value instead of this self-updated one
+	highFreqKHZ  int
+	effectKeeper effectKeeper
+
+	cpuFreqReader reader.MetricReader
+}
+
+type effectKeeper struct {
+	effectiveValue int
+}
+
+func (h *effectKeeper) Update(curr int) int {
+	// not to smooth out incoming maxima yet; should it be spiky, it shall be averaged out eventually
+	if curr > h.effectiveValue {
+		h.effectiveValue = curr
+		return curr
+	}
+
+	h.effectiveValue = int(ema(float64(h.effectiveValue), float64(curr), alpha))
+	return h.effectiveValue
+}
+
+// ema implements Sliding-Window Exponential Moving Average algorithm
+func ema(meanAverage, value, alpha float64) float64 {
+	if meanAverage <= 0 {
+		return value
+	}
+
+	return meanAverage*(1-alpha) + value*alpha
+}
+
+func (h *effectKeeper) Clear() {
+	h.effectiveValue = 0
+}
+
+func (c *cpuFreqChangeAssessor) IsInitialized() bool {
+	return c.highFreqKHZ >= minKHZ
+}
+
+func (c *cpuFreqChangeAssessor) Init() error {
+	freq, err := c.cpuFreqReader.Get(context.Background())
+	if err != nil {
+		return errors.Wrap(err, "failed to get initial cpu freq value")
+	}
+
+	general.Infof("pap: cpufreq set initial value %d khz", freq)
+	c.highFreqKHZ = freq
+	return nil
+}
+
+func (c *cpuFreqChangeAssessor) Clear() {
+	c.effectKeeper.Clear()
+}
+
+func (c *cpuFreqChangeAssessor) AssessEffect(_ int, _, _ bool) (int, error) {
+	// always check cpu freq to assess the effect
+	currentFreq, err := c.cpuFreqReader.Get(context.Background())
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to fetch latest cpu freq to access dvfs effect")
+	}
+
+	if currentFreq > c.highFreqKHZ {
+		c.highFreqKHZ = currentFreq
+		c.effectKeeper.Clear()
+	}
+
+	return c.assessEffectByFreq(currentFreq)
+}
+
+func (c *cpuFreqChangeAssessor) assessEffectByFreq(currentFreq int) (int, error) {
+	general.InfofV(6, "pap: cpuFreqChangeAssessor assessEffectByFreq: curr %d, base %d", currentFreq, c.highFreqKHZ)
+	if currentFreq < minKHZ {
+		return 0, fmt.Errorf("invalid currentFreq frequency %d khz", currentFreq)
+	}
+
+	if currentFreq >= c.highFreqKHZ {
+		return 0, nil
+	}
+
+	instantEffect := 100 - currentFreq*100/c.highFreqKHZ
+	return c.effectKeeper.Update(instantEffect), nil
+}
+
+func (c *cpuFreqChangeAssessor) Update(_ int) {
+	// no need to keep track of the cpu freq change as the initial cpu freq is always the baseline
+}
+
+func (c *cpuFreqChangeAssessor) AssessTarget(actualWatt, desiredWatt int, maxDecreasePercent int) int {
+	// keep as is if there is no room to decrease
+	if maxDecreasePercent <= 0 {
+		return actualWatt
+	}
+
+	// when there is decrease room for cpu frequency, lower the power in smaller portion to avoid misstep
+	lowerLimit := (100 - maxDecreasePercent/2) * actualWatt / 100
+	if lowerLimit > desiredWatt {
+		return lowerLimit
+	}
+
+	return desiredWatt
+}
+
+func NewCPUFreqChangeAssessor(initKHZ int, nodeMetricGetter reader.NodeMetricGetter) Assessor {
+	return &cpuFreqChangeAssessor{
+		highFreqKHZ:   initKHZ,
+		effectKeeper:  effectKeeper{},
+		cpuFreqReader: reader.NewCPUFreqReader(nodeMetricGetter),
+	}
+}

pkg/agent/sysadvisor/plugin/poweraware/advisor/action/strategy/assess/cpufreq_change_test.go

@@ -0,0 +1,181 @@
+/*
+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 assess
+
+import "testing"
+
+func Test_cpuFreqChangeAssessor_AccumulateEffect(t *testing.T) {
+	t.Parallel()
+	type fields struct {
+		initFreqKHZ int
+	}
+	type args struct {
+		current int
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		want    int
+		wantErr bool
+	}{
+		{
+			name: "happy path",
+			fields: fields{
+				initFreqKHZ: 2500_000,
+			},
+			args: args{
+				current: 2000_000,
+			},
+			want:    20,
+			wantErr: false,
+		},
+		{
+			name: "negative low current freq",
+			fields: fields{
+				initFreqKHZ: 2500_000,
+			},
+			args: args{
+				current: 777_000,
+			},
+			want:    0,
+			wantErr: true,
+		},
+		{
+			name: "fine for higher freq no change as temporary spike",
+			fields: fields{
+				initFreqKHZ: 2500_000,
+			},
+			args: args{
+				current: 2600_000,
+			},
+			want:    0,
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			c := &cpuFreqChangeAssessor{
+				highFreqKHZ: tt.fields.initFreqKHZ,
+			}
+			got, err := c.assessEffectByFreq(tt.args.current)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("AssessEffect() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("AssessEffect() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_cpuFreqChangeAssessor_AssessTarget(t *testing.T) {
+	t.Parallel()
+	type fields struct {
+		initFreqMhz int
+	}
+	type args struct {
+		actualWatt         int
+		desiredWatt        int
+		maxDecreasePercent int
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   int
+	}{
+		{
+			name: "happy path",
+			fields: fields{
+				initFreqMhz: 2500,
+			},
+			args: args{
+				actualWatt:         100,
+				desiredWatt:        80,
+				maxDecreasePercent: 10,
+			},
+			want: 95,
+		},
+		{
+			name: "happy path",
+			fields: fields{
+				initFreqMhz: 2500,
+			},
+			args: args{
+				actualWatt:         100,
+				desiredWatt:        80,
+				maxDecreasePercent: 0,
+			},
+			want: 100,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			c := &cpuFreqChangeAssessor{
+				highFreqKHZ: tt.fields.initFreqMhz,
+			}
+			if got := c.AssessTarget(tt.args.actualWatt, tt.args.desiredWatt, tt.args.maxDecreasePercent); got != tt.want {
+				t.Errorf("AssessTarget() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_effectKeeper_Update(t *testing.T) {
+	t.Parallel()
+	type fields struct {
+		value int
+	}
+	type args struct {
+		curr int
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   int
+	}{
+		{
+			name: "happy path to smooth out old spike",
+			fields: fields{
+				value: 100,
+			},
+			args: args{
+				curr: 80,
+			},
+			want: 92,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			h := &effectKeeper{
+				effectiveValue: tt.fields.value,
+			}
+			if got := h.Update(tt.args.curr); got != tt.want {
+				t.Errorf("Update() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}