observe metrics in recommendation controller

cmd/craned/app/manager.go

@@ -368,11 +368,12 @@ func initControllers(oomRecorder oom.Recorder, mgr ctrl.Manager, opts *options.O
 		}
 
 		if err := (&recommendationctrl.RecommendationController{
-			Client:      mgr.GetClient(),
-			Scheme:      mgr.GetScheme(),
-			RestMapper:  mgr.GetRESTMapper(),
-			ScaleClient: scaleClient,
-			Recorder:    mgr.GetEventRecorderFor("recommendation-controller"),
+			Client:         mgr.GetClient(),
+			Scheme:         mgr.GetScheme(),
+			RestMapper:     mgr.GetRESTMapper(),
+			RecommenderMgr: recommenderMgr,
+			ScaleClient:    scaleClient,
+			Recorder:       mgr.GetEventRecorderFor("recommendation-controller"),
 		}).SetupWithManager(mgr); err != nil {
 			klog.Exit(err, "unable to create controller", "controller", "RecommendationController")
 		}

go.mod

@@ -29,6 +29,7 @@ require (
 	k8s.io/kubelet v0.22.3
 	k8s.io/kubernetes v1.22.3
 	k8s.io/metrics v0.22.3
+	k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a
 	sigs.k8s.io/controller-runtime v0.10.2
 	sigs.k8s.io/custom-metrics-apiserver v1.22.0
 	sigs.k8s.io/prometheus-adapter v0.9.0
@@ -163,7 +164,6 @@ require (
 	k8s.io/component-helpers v0.22.3 // indirect
 	k8s.io/kube-scheduler v0.0.0 // indirect
 	k8s.io/mount-utils v0.22.3 // indirect
-	k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.22 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.1.2 // indirect
 )

pkg/controller/recommendation/recommendation_controller.go

@@ -4,15 +4,14 @@ import (
 	"context"
 	"fmt"
 
-	"k8s.io/klog/v2"
-
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/scale"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -21,18 +20,21 @@ import (
 	analysisv1alph1 "github.com/gocrane/api/analysis/v1alpha1"
 	predictormgr "github.com/gocrane/crane/pkg/predictor"
 	"github.com/gocrane/crane/pkg/providers"
+	recommender "github.com/gocrane/crane/pkg/recommendation"
+	"github.com/gocrane/crane/pkg/recommendation/framework"
 )
 
 // RecommendationController is responsible for reconcile Recommendation
 type RecommendationController struct {
 	client.Client
-	ConfigSet    *analysisv1alph1.ConfigSet
-	Scheme       *runtime.Scheme
-	Recorder     record.EventRecorder
-	RestMapper   meta.RESTMapper
-	ScaleClient  scale.ScalesGetter
-	PredictorMgr predictormgr.Manager
-	Provider     providers.History
+	ConfigSet      *analysisv1alph1.ConfigSet
+	Scheme         *runtime.Scheme
+	Recorder       record.EventRecorder
+	RestMapper     meta.RESTMapper
+	RecommenderMgr recommender.RecommenderManager
+	ScaleClient    scale.ScalesGetter
+	PredictorMgr   predictormgr.Manager
+	Provider       providers.History
 }
 
 func (c *RecommendationController) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -48,6 +50,17 @@ func (c *RecommendationController) Reconcile(ctx context.Context, req ctrl.Reque
 		return ctrl.Result{}, nil
 	}
 
+	r, err := c.RecommenderMgr.GetRecommender(string(recommendation.Spec.Type))
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	recommendationContext := framework.NewRecommendationContextForObserve(recommendation, c.RestMapper, c.ScaleClient)
+	err = r.Observe(&recommendationContext)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
 	// defaulting for TargetRef.Namespace
 	if recommendation.Spec.TargetRef.Namespace == "" {
 		recommendation.Spec.TargetRef.Namespace = recommendation.Namespace

pkg/controller/recommendation/recommendation_rule_controller.go

@@ -363,7 +363,7 @@ func (c *RecommendationRuleController) executeMission(ctx context.Context, wg *s
 			recommendation = c.CreateRecommendationObject(recommendationRule, mission.TargetRef, id, mission.RecommenderRef.Name)
 		}
 
-		r, err := c.RecommenderMgr.GetRecommender(mission.RecommenderRef.Name, *recommendationRule)
+		r, err := c.RecommenderMgr.GetRecommenderWithRule(mission.RecommenderRef.Name, *recommendationRule)
 		if err != nil {
 			mission.Message = fmt.Sprintf("get recommender %s failed, %v", mission.RecommenderRef.Name, err)
 			return

pkg/recommendation/framework/context.go

@@ -87,6 +87,14 @@ func NewRecommendationContext(context context.Context, identity ObjectIdentity,
 	}
 }
 
+func NewRecommendationContextForObserve(recommendation *v1alpha1.Recommendation, restMapper meta.RESTMapper, scaleClient scale.ScalesGetter) RecommendationContext {
+	return RecommendationContext{
+		Recommendation: recommendation,
+		RestMapper:     restMapper,
+		ScaleClient:    scaleClient,
+	}
+}
+
 func (ctx RecommendationContext) String() string {
 	return fmt.Sprintf("RecommendationRule(%s) Target(%s/%s)", ctx.RecommendationRule.Name, ctx.Object.GetNamespace(), ctx.Object.GetName())
 }

pkg/recommendation/manager.go

@@ -22,7 +22,9 @@ import (
 
 type RecommenderManager interface {
 	// GetRecommender return a registered recommender
-	GetRecommender(recommenderName string, recommendationRule analysisv1alph1.RecommendationRule) (recommender.Recommender, error)
+	GetRecommender(recommenderName string) (recommender.Recommender, error)
+	// GetRecommenderWithRule return a registered recommender, its config merged with recommendationRule
+	GetRecommenderWithRule(recommenderName string, recommendationRule analysisv1alph1.RecommendationRule) (recommender.Recommender, error)
 }
 
 func NewRecommenderManager(recommendationConfiguration string, oomRecorder oom.Recorder, realtimeDataSources map[providers.DataSourceType]providers.RealTime, historyDataSources map[providers.DataSourceType]providers.History) RecommenderManager {
@@ -53,7 +55,11 @@ type manager struct {
 	oomRecorder        oom.Recorder
 }
 
-func (m *manager) GetRecommender(recommenderName string, recommendationRule analysisv1alph1.RecommendationRule) (recommender.Recommender, error) {
+func (m *manager) GetRecommender(recommenderName string) (recommender.Recommender, error) {
+	return m.GetRecommenderWithRule(recommenderName, analysisv1alph1.RecommendationRule{})
+}
+
+func (m *manager) GetRecommenderWithRule(recommenderName string, recommendationRule analysisv1alph1.RecommendationRule) (recommender.Recommender, error) {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 

pkg/recommendation/recommender/resource/observe.go

@@ -1,6 +1,7 @@
 package resource
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 
@@ -10,6 +11,7 @@ import (
 
 	"github.com/gocrane/crane/pkg/metrics"
 	"github.com/gocrane/crane/pkg/recommendation/framework"
+	"github.com/gocrane/crane/pkg/utils"
 )
 
 // Observe enhance the observability.
@@ -32,14 +34,20 @@ func (rr *ResourceRecommender) Observe(ctx *framework.RecommendationContext) err
 	}
 
 	for _, container := range newPodTemplate.Spec.Containers {
-		rr.recordResourceRecommendation(ctx, container.Name, v1.ResourceCPU, container.Resources.Requests[v1.ResourceCPU])
+		err = rr.recordResourceRecommendation(ctx, container.Name, v1.ResourceCPU, container.Resources.Requests[v1.ResourceCPU])
+		if err != nil {
+			return err
+		}
 		rr.recordResourceRecommendation(ctx, container.Name, v1.ResourceMemory, container.Resources.Requests[v1.ResourceMemory])
+		if err != nil {
+			return err
+		}
 	}
 
 	return nil
 }
 
-func (rr *ResourceRecommender) recordResourceRecommendation(ctx *framework.RecommendationContext, containerName string, resName v1.ResourceName, quantity resource.Quantity) {
+func (rr *ResourceRecommender) recordResourceRecommendation(ctx *framework.RecommendationContext, containerName string, resName v1.ResourceName, quantity resource.Quantity) error {
 	labels := map[string]string{
 		"apiversion": ctx.Recommendation.Spec.TargetRef.APIVersion,
 		"owner_kind": ctx.Recommendation.Spec.TargetRef.Kind,
@@ -49,12 +57,18 @@ func (rr *ResourceRecommender) recordResourceRecommendation(ctx *framework.Recom
 		"resource":   resName.String(),
 	}
 
-	labels["owner_replicas"] = fmt.Sprintf("%d", len(ctx.Pods))
+	scale, _, err := utils.GetScaleFromObjectReference(context.TODO(), ctx.RestMapper, ctx.ScaleClient, ctx.Recommendation.Spec.TargetRef)
+	if err != nil {
+		return err
+	}
+	labels["owner_replicas"] = fmt.Sprintf("%d", scale.Spec.Replicas)
 
 	switch resName {
 	case v1.ResourceCPU:
 		metrics.ResourceRecommendation.With(labels).Set(float64(quantity.MilliValue()) / 1000.)
 	case v1.ResourceMemory:
 		metrics.ResourceRecommendation.With(labels).Set(float64(quantity.Value()))
 	}
+
+	return nil
 }

pkg/utils/scale.go

@@ -44,6 +44,15 @@ func GetScale(ctx context.Context, restMapper meta.RESTMapper, scaleClient scale
 	return nil, nil, fmt.Errorf("unrecognized resource: %+v", errs)
 }
 
+func GetScaleFromObjectReference(ctx context.Context, restMapper meta.RESTMapper, scaleClient scale.ScalesGetter, ref v1.ObjectReference) (*autoscalingapiv1.Scale, *meta.RESTMapping, error) {
+	newRef := autoscalingv2.CrossVersionObjectReference{
+		APIVersion: ref.APIVersion,
+		Kind:       ref.Kind,
+		Name:       ref.Name,
+	}
+	return GetScale(ctx, restMapper, scaleClient, ref.Namespace, newRef)
+}
+
 func GetPodsFromScale(kubeClient client.Client, scale *autoscalingapiv1.Scale) ([]v1.Pod, error) {
 	selector, err := labels.ConvertSelectorToLabelsMap(scale.Status.Selector)
 	if err != nil {