Merge pull request #775 from lbbniu/feat/lbbniu/idlenode

feat(idlenode): Idle Node Plugin Enhancement

pkg/metricnaming/naming.go

@@ -9,14 +9,14 @@ import (
 
 // MetricNamer is an interface. it is the bridge between predictor and different data sources and other component such as caller.
 type MetricNamer interface {
-	// Used for datasource provider, data source provider call QueryBuilder
+	// QueryBuilder used for datasource provider, data source provider call QueryBuilder
 	QueryBuilder() querybuilder.QueryBuilder
-	// Used for predictor now
+	// BuildUniqueKey used for predictor now
 	BuildUniqueKey() string
 
 	Validate() error
 
-	// Means the caller of this MetricNamer, different caller maybe use the same metric
+	// Caller means the caller of this MetricNamer, different caller maybe use the same metric
 	Caller() string
 }
 
@@ -94,3 +94,18 @@ func ResourceToContainerMetricNamer(namespace, apiVersion, workloadKind, workloa
 		},
 	}
 }
+
+func ResourceToGeneralMetricNamer(queryExpr string, resourceName corev1.ResourceName, nodeLabelSelector labels.Selector, caller string) MetricNamer {
+	// node
+	return &GeneralMetricNamer{
+		CallerName: caller,
+		Metric: &metricquery.Metric{
+			Type:       metricquery.PromQLMetricType,
+			MetricName: resourceName.String(),
+			Prom: &metricquery.PromNamerInfo{
+				QueryExpr: queryExpr,
+				Selector:  nodeLabelSelector,
+			},
+		},
+	}
+}

pkg/prometheus-adapter/config_fetcher.go

@@ -78,7 +78,7 @@ func (pc *PrometheusAdapterConfigFetcher) SetupWithManager(mgr ctrl.Manager) err
 		Complete(pc)
 }
 
-// fetched metricRule if configmap is updated
+// Update fetched metricRule if configmap is updated
 func (paCm *PrometheusAdapterConfigChangedPredicate) Update(e event.UpdateEvent) bool {
 	if e.ObjectOld == nil {
 		return false

pkg/recommendation/config/config.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"io/ioutil"
 
-	klog "k8s.io/klog/v2"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/yaml"
 
 	analysisv1alph1 "github.com/gocrane/api/analysis/v1alpha1"

pkg/recommendation/framework/context.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sync"
 
 	jsonpatch "github.com/evanphx/json-patch"
 	appsv1 "k8s.io/api/apps/v1"
@@ -33,10 +34,10 @@ type RecommendationContext struct {
 	Identity ObjectIdentity
 	// Target Object
 	Object client.Object
+	// Protecting inputValues
+	inputValuesMutex sync.RWMutex
 	// Time series data from data source.
-	InputValues []*common.TimeSeries
-	// Time series data 2 from data source.
-	InputValues2 []*common.TimeSeries
+	inputValues map[string][]*common.TimeSeries
 	// Result series from prediction
 	ResultValues []*common.TimeSeries
 	// DataProviders contains data source of your recommendation flow.
@@ -73,29 +74,42 @@ type RecommendationContext struct {
 
 func NewRecommendationContext(context context.Context, identity ObjectIdentity, recommendationRule *v1alpha1.RecommendationRule, predictorMgr predictormgr.Manager, dataProviders map[providers.DataSourceType]providers.History, recommendation *v1alpha1.Recommendation, client client.Client, scaleClient scale.ScalesGetter) RecommendationContext {
 	return RecommendationContext{
+		Context:            context,
 		Identity:           identity,
 		Object:             &identity.Object,
-		Context:            context,
+		inputValues:        make(map[string][]*common.TimeSeries),
 		PredictorMgr:       predictorMgr,
 		DataProviders:      dataProviders,
 		RecommendationRule: recommendationRule,
 		Recommendation:     recommendation,
 		Client:             client,
 		RestMapper:         client.RESTMapper(),
 		ScaleClient:        scaleClient,
-		//CancelCh:       context.Done(),
 	}
 }
 
 func NewRecommendationContextForObserve(recommendation *v1alpha1.Recommendation, restMapper meta.RESTMapper, scaleClient scale.ScalesGetter) RecommendationContext {
 	return RecommendationContext{
+		inputValues:    make(map[string][]*common.TimeSeries),
 		Recommendation: recommendation,
 		RestMapper:     restMapper,
 		ScaleClient:    scaleClient,
 	}
 }
 
-func (ctx RecommendationContext) String() string {
+func (ctx *RecommendationContext) AddInputValue(key string, timeSeries []*common.TimeSeries) {
+	ctx.inputValuesMutex.Lock()
+	defer ctx.inputValuesMutex.Unlock()
+	ctx.inputValues[key] = timeSeries
+}
+
+func (ctx *RecommendationContext) InputValue(key string) []*common.TimeSeries {
+	ctx.inputValuesMutex.RLock()
+	defer ctx.inputValuesMutex.RUnlock()
+	return ctx.inputValues[key]
+}
+
+func (ctx *RecommendationContext) String() string {
 	return fmt.Sprintf("RecommendationRule(%s) Target(%s/%s)", ctx.RecommendationRule.Name, ctx.Object.GetNamespace(), ctx.Object.GetName())
 }
 

pkg/recommendation/recommender/apis/recommender.go

@@ -0,0 +1,41 @@
+package apis
+
+import (
+	"strconv"
+	"time"
+)
+
+func (r Recommender) GetConfigFloat(key string, def float64) (float64, error) {
+	if value, exists := r.Config[key]; exists {
+		return strconv.ParseFloat(value, 64)
+	}
+	return def, nil
+}
+
+func (r Recommender) GetConfigInt(key string, def int64) (int64, error) {
+	if value, exists := r.Config[key]; exists {
+		return strconv.ParseInt(value, 10, 64)
+	}
+	return def, nil
+}
+
+func (r Recommender) GetConfigBool(key string, def bool) (bool, error) {
+	if value, exists := r.Config[key]; exists {
+		return strconv.ParseBool(value)
+	}
+	return def, nil
+}
+
+func (r Recommender) GetConfigString(key string, def string) string {
+	if value, exists := r.Config[key]; exists {
+		return value
+	}
+	return def
+}
+
+func (r Recommender) GetConfigDuration(key string, def time.Duration) (time.Duration, error) {
+	if value, exists := r.Config[key]; exists {
+		return time.ParseDuration(value)
+	}
+	return def, nil
+}

pkg/recommendation/recommender/hpa/recommend.go

@@ -58,7 +58,7 @@ func (rr *HPARecommender) Policy(ctx *framework.RecommendationContext) error {
 		return fmt.Errorf("checkMinCpuUsageThreshold failed: %v", err)
 	}
 
-	medianMin, medianMax, err := rr.minMaxMedians(ctx.InputValues)
+	medianMin, medianMax, err := rr.minMaxMedians(ctx.InputValue(string(corev1.ResourceCPU)))
 	if err != nil {
 		return fmt.Errorf("minMaxMedians failed: %v", err)
 	}

pkg/recommendation/recommender/hpa/registry.go

@@ -1,8 +1,6 @@
 package hpa
 
 import (
-	"strconv"
-
 	analysisv1alph1 "github.com/gocrane/api/analysis/v1alpha1"
 	"github.com/gocrane/crane/pkg/recommendation/config"
 	"github.com/gocrane/crane/pkg/recommendation/recommender"
@@ -31,65 +29,37 @@ func (rr *HPARecommender) Name() string {
 func NewHPARecommender(recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (*HPARecommender, error) {
 	recommender = config.MergeRecommenderConfigFromRule(recommender, recommendationRule)
 
-	predictable, exists := recommender.Config["predictable"]
-	if !exists {
-		predictable = "false"
-	}
-	predictableEnabled, err := strconv.ParseBool(predictable)
+	predictableEnabled, err := recommender.GetConfigBool("predictable", false)
 	if err != nil {
 		return nil, err
 	}
 
-	referenceHPA, exists := recommender.Config["reference-hpa"]
-	if !exists {
-		referenceHPA = "true"
-	}
-	referenceHpaEnabled, err := strconv.ParseBool(referenceHPA)
+	referenceHpaEnabled, err := recommender.GetConfigBool("reference-hpa", true)
 	if err != nil {
 		return nil, err
 	}
 
-	minCpuUsageThreshold, exists := recommender.Config["min-cpu-usage-threshold"]
-	if !exists {
-		minCpuUsageThreshold = "1"
-	}
-	minCpuUsageThresholdFloat, err := strconv.ParseFloat(minCpuUsageThreshold, 64)
+	minCpuUsageThresholdFloat, err := recommender.GetConfigFloat("min-cpu-usage-threshold", 1)
 	if err != nil {
 		return nil, err
 	}
 
-	fluctuationThreshold, exists := recommender.Config["fluctuation-threshold"]
-	if !exists {
-		fluctuationThreshold = "1.5"
-	}
-	fluctuationThresholdFloat, err := strconv.ParseFloat(fluctuationThreshold, 64)
+	fluctuationThresholdFloat, err := recommender.GetConfigFloat("fluctuation-threshold", 1.5)
 	if err != nil {
 		return nil, err
 	}
 
-	minCpuTargetUtilization, exists := recommender.Config["min-cpu-target-utilization"]
-	if !exists {
-		minCpuTargetUtilization = "30"
-	}
-	minCpuTargetUtilizationInt, err := strconv.ParseInt(minCpuTargetUtilization, 10, 32)
+	minCpuTargetUtilizationInt, err := recommender.GetConfigInt("min-cpu-target-utilization", 30)
 	if err != nil {
 		return nil, err
 	}
 
-	maxCpuTargetUtilization, exists := recommender.Config["max-cpu-target-utilization"]
-	if !exists {
-		maxCpuTargetUtilization = "75"
-	}
-	maxCpuTargetUtilizationInt, err := strconv.ParseInt(maxCpuTargetUtilization, 10, 32)
+	maxCpuTargetUtilizationInt, err := recommender.GetConfigInt("max-cpu-target-utilization", 75)
 	if err != nil {
 		return nil, err
 	}
 
-	maxReplicasFactor, exists := recommender.Config["max-replicas-factor"]
-	if !exists {
-		maxReplicasFactor = "3"
-	}
-	maxReplicasFactorFloat, err := strconv.ParseFloat(maxReplicasFactor, 64)
+	maxReplicasFactorFloat, err := recommender.GetConfigFloat("max-replicas-factor", 3)
 	if err != nil {
 		return nil, err
 	}

pkg/recommendation/recommender/idlenode/prepare.go

@@ -1,9 +1,21 @@
 package idlenode
 
 import (
+	"fmt"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/klog/v2"
+
+	"github.com/gocrane/crane/pkg/metricnaming"
+	"github.com/gocrane/crane/pkg/providers"
 	"github.com/gocrane/crane/pkg/recommendation/framework"
+	"github.com/gocrane/crane/pkg/utils"
 )
 
+const callerFormat = "IdleNodeRecommender-%s-%s"
+
 // CheckDataProviders in PrePrepare phase, will create data source provider via your recommendation config.
 func (inr *IdleNodeRecommender) CheckDataProviders(ctx *framework.RecommendationContext) error {
 	if err := inr.BaseRecommender.CheckDataProviders(ctx); err != nil {
@@ -14,6 +26,82 @@ func (inr *IdleNodeRecommender) CheckDataProviders(ctx *framework.Recommendation
 }
 
 func (inr *IdleNodeRecommender) CollectData(ctx *framework.RecommendationContext) error {
+	labelSelector := labels.SelectorFromSet(ctx.Identity.Labels)
+	caller := fmt.Sprintf(callerFormat, klog.KObj(ctx.Recommendation), ctx.Recommendation.UID)
+	timeNow := time.Now()
+	if inr.cpuUsageUtilization > 0 {
+		metricNamer := metricnaming.ResourceToGeneralMetricNamer(utils.GetNodeCpuUsageUtilizationExpression(ctx.Recommendation.Spec.TargetRef.Name), corev1.ResourceCPU, labelSelector, caller)
+		if err := metricNamer.Validate(); err != nil {
+			return err
+		}
+		ctx.MetricNamer = metricNamer
+
+		// get node cpu usage utilization
+		klog.Infof("%s: %s CpuQuery %s", ctx.String(), inr.Name(), ctx.MetricNamer.BuildUniqueKey())
+		tsList, err := ctx.DataProviders[providers.PrometheusDataSource].QueryTimeSeries(ctx.MetricNamer, timeNow.Add(-time.Hour*24*7), timeNow, time.Minute)
+		if err != nil {
+			return fmt.Errorf("%s query node cpu usage historic metrics failed: %v ", inr.Name(), err)
+		}
+		if len(tsList) != 1 {
+			return fmt.Errorf("%s query node cpu usage historic metrics data is unexpected, List length is %d ", inr.Name(), len(tsList))
+		}
+		ctx.AddInputValue(cpuUsageUtilizationKey, tsList)
+	}
+
+	if inr.memoryUsageUtilization > 0 {
+		metricNamer := metricnaming.ResourceToGeneralMetricNamer(utils.GetNodeMemUsageUtilizationExpression(ctx.Recommendation.Spec.TargetRef.Name), corev1.ResourceMemory, labelSelector, caller)
+		if err := metricNamer.Validate(); err != nil {
+			return err
+		}
+		// get node memory usage utilization
+		klog.Infof("%s: %s MemoryQuery %s", ctx.String(), inr.Name(), metricNamer.BuildUniqueKey())
+		tsList, err := ctx.DataProviders[providers.PrometheusDataSource].QueryTimeSeries(metricNamer, timeNow.Add(-time.Hour*24*7), timeNow, time.Minute)
+		if err != nil {
+			return fmt.Errorf("%s query node memory usage historic metrics failed: %v ", inr.Name(), err)
+		}
+		if len(tsList) != 1 {
+			return fmt.Errorf("%s query node memory usage historic metrics data is unexpected, List length is %d ", inr.Name(), len(tsList))
+		}
+		ctx.AddInputValue(memoryUsageUtilizationKey, tsList)
+	}
+
+	if inr.cpuRequestUtilization > 0 {
+		metricNamer := metricnaming.ResourceToGeneralMetricNamer(utils.GetNodeCpuRequestUtilizationExpression(ctx.Recommendation.Spec.TargetRef.Name), corev1.ResourceCPU, labelSelector, caller)
+		if err := metricNamer.Validate(); err != nil {
+			return err
+		}
+		ctx.MetricNamer = metricNamer
+
+		// get node cpu request utilization
+		klog.Infof("%s: %s CpuQuery %s", ctx.String(), inr.Name(), metricNamer)
+		tsList, err := ctx.DataProviders[providers.PrometheusDataSource].QueryTimeSeries(metricNamer, timeNow.Add(-time.Hour*24*7), timeNow, time.Minute)
+		if err != nil {
+			return fmt.Errorf("%s query node cpu request historic metrics failed: %v ", inr.Name(), err)
+		}
+		if len(tsList) != 1 {
+			return fmt.Errorf("%s query node cpu request historic metrics data is unexpected, List length is %d ", inr.Name(), len(tsList))
+		}
+		ctx.AddInputValue(cpuRequestUtilizationKey, tsList)
+	}
+
+	if inr.memoryRequestUtilization > 0 {
+		metricNamer := metricnaming.ResourceToGeneralMetricNamer(utils.GetNodeMemRequestUtilizationExpression(ctx.Recommendation.Spec.TargetRef.Name), corev1.ResourceMemory, labelSelector, caller)
+		if err := metricNamer.Validate(); err != nil {
+			return err
+		}
+
+		// get node memory request utilization
+		klog.Infof("%s: %s MemoryQuery %s", ctx.String(), inr.Name(), metricNamer.BuildUniqueKey())
+		tsList, err := ctx.DataProviders[providers.PrometheusDataSource].QueryTimeSeries(metricNamer, timeNow.Add(-time.Hour*24*7), timeNow, time.Minute)
+		if err != nil {
+			return fmt.Errorf("%s query node memory request historic metrics failed: %v ", inr.Name(), err)
+		}
+		if len(tsList) != 1 {
+			return fmt.Errorf("%s query node memory request historic metrics data is unexpected, List length is %d ", inr.Name(), len(tsList))
+		}
+		ctx.AddInputValue(memoryRequestUtilizationKey, tsList)
+	}
+
 	return nil
 }