pap: power reader uses power metrics served by metric store

pkg/agent/sysadvisor/plugin/poweraware/advisor/advisor.go

@@ -36,8 +36,9 @@ import (
 )
 
 const (
-	// 8 seconds between actions since RAPL/HSMP capping needs 4-6 seconds to stabilize itself
-	intervalSpecFetch = time.Second * 8
+	// 9 seconds between actions since RAPL/HSMP capping needs 4-6 seconds to stabilize itself
+	// and malachite realtime metric server imposes delay of up to 2 seconds
+	intervalSpecFetch = time.Second * 9
 
 	metricPowerAwareCurrentPowerInWatt = "power_current_watt"
 	metricPowerAwareDesiredPowerInWatt = "power_desired_watt"

pkg/agent/sysadvisor/plugin/poweraware/power_aware.go

@@ -94,9 +94,7 @@ func NewPowerAwarePlugin(
 		}
 	}
 
-	// todo: use the power usage data from malachite
-	// we may temporarily have a local reader on top of ipmi (in dev branch), before malachite is ready
-	powerReader := reader.NewDummyPowerReader()
+	powerReader := reader.NewMetricStorePowerReader(metaServer)
 
 	powerAdvisor := advisor.NewAdvisor(conf.PowerAwarePluginConfiguration.DryRun,
 		conf.PowerAwarePluginConfiguration.AnnotationKeyPrefix,

pkg/agent/sysadvisor/plugin/poweraware/reader/metric_store_power_reader.go

@@ -0,0 +1,74 @@
+/*
+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 reader
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/kubewharf/katalyst-core/pkg/consts"
+	utilmetric "github.com/kubewharf/katalyst-core/pkg/util/metric"
+)
+
+// malachite realtime power metric server imposes delay of up to 2 seconds coupled with sampling interval of 1 sec
+const powerTolerationTime = 3 * time.Second
+
+type nodeMetricGetter interface {
+	GetNodeMetric(metricName string) (utilmetric.MetricData, error)
+}
+
+type metricStorePowerReader struct {
+	nodeMetricGetter
+}
+
+func (m *metricStorePowerReader) Init() error {
+	return nil
+}
+
+func isDataFresh(data utilmetric.MetricData, now time.Time) bool {
+	if data.Time == nil {
+		return false
+	}
+	return now.Before(data.Time.Add(powerTolerationTime))
+}
+
+func (m *metricStorePowerReader) Get(ctx context.Context) (int, error) {
+	return m.get(ctx, time.Now())
+}
+
+func (m *metricStorePowerReader) get(ctx context.Context, now time.Time) (int, error) {
+	data, err := m.GetNodeMetric(consts.MetricTotalPowerUsedWatts)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to get metric from metric store")
+	}
+
+	if !isDataFresh(data, now) {
+		return 0, errors.New("power data in metric store is stale")
+	}
+
+	return int(data.Value), nil
+}
+
+func (m *metricStorePowerReader) Cleanup() {}
+
+func NewMetricStorePowerReader(nodeMetricGetter nodeMetricGetter) PowerReader {
+	return &metricStorePowerReader{
+		nodeMetricGetter: nodeMetricGetter,
+	}
+}

pkg/agent/sysadvisor/plugin/poweraware/reader/metric_store_power_reader_test.go

@@ -0,0 +1,103 @@
+/*
+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 reader
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	utilmetric "github.com/kubewharf/katalyst-core/pkg/util/metric"
+)
+
+func Test_metricStorePowerReader_get(t *testing.T) {
+	t.Parallel()
+
+	setTime := time.Date(2024, 11, 11, 8, 30, 10, 0, time.UTC)
+	dummyMetricStore := utilmetric.NewMetricStore()
+	dummyMetricStore.SetNodeMetric("total.power.used.watts", utilmetric.MetricData{
+		Value: 999,
+		Time:  &setTime,
+	})
+
+	type fields struct {
+		metricStore *utilmetric.MetricStore
+	}
+	type args struct {
+		ctx context.Context
+		now time.Time
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		want    int
+		wantErr bool
+	}{
+		{
+			name: "happy path",
+			fields: fields{
+				metricStore: dummyMetricStore,
+			},
+			args: args{
+				ctx: context.TODO(),
+				now: time.Date(2024, 11, 11, 8, 30, 11, 0, time.UTC),
+			},
+			want:    999,
+			wantErr: false,
+		},
+		{
+			name: "stale data",
+			fields: fields{
+				metricStore: dummyMetricStore,
+			},
+			args: args{
+				ctx: context.TODO(),
+				now: time.Date(2024, 11, 11, 8, 30, 13, 0, time.UTC),
+			},
+			want:    0,
+			wantErr: true,
+		},
+		{
+			name: "no such metric in store",
+			fields: fields{
+				metricStore: utilmetric.NewMetricStore(),
+			},
+			args: args{
+				ctx: context.TODO(),
+				now: time.Date(2024, 11, 11, 8, 30, 13, 0, time.UTC),
+			},
+			want:    0,
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			m := NewMetricStorePowerReader(tt.fields.metricStore).(*metricStorePowerReader)
+			got, err := m.get(tt.args.ctx, tt.args.now)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("get() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("get() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}