Merge pull request gocrane#785 from lbbniu/perf/lbbniu/recommender

perf: optimize recommender manager

cmd/craned/app/manager.go

@@ -138,7 +138,7 @@ func Run(ctx context.Context, opts *options.Options) error {
 		}
 	}()
 
-	recommenderMgr := initRecommenderManager(opts, podOOMRecorder, realtimeDataSources, historyDataSources)
+	recommenderMgr := initRecommenderManager(opts)
 	initControllers(podOOMRecorder, mgr, opts, predictorMgr, recommenderMgr, historyDataSources[providers.PrometheusDataSource])
 	// initialize custom collector metrics
 	initMetricCollector(mgr)
@@ -147,8 +147,8 @@ func Run(ctx context.Context, opts *options.Options) error {
 	return nil
 }
 
-func initRecommenderManager(opts *options.Options, oomRecorder oom.Recorder, realtimeDataSources map[providers.DataSourceType]providers.RealTime, historyDataSources map[providers.DataSourceType]providers.History) recommendation.RecommenderManager {
-	return recommendation.NewRecommenderManager(opts.RecommendationConfiguration, oomRecorder, realtimeDataSources, historyDataSources)
+func initRecommenderManager(opts *options.Options) recommendation.RecommenderManager {
+	return recommendation.NewRecommenderManager(opts.RecommendationConfiguration)
 }
 
 func initScheme() {
@@ -392,6 +392,7 @@ func initControllers(oomRecorder oom.Recorder, mgr ctrl.Manager, opts *options.O
 			RestMapper:     mgr.GetRESTMapper(),
 			RecommenderMgr: recommenderMgr,
 			ScaleClient:    scaleClient,
+			OOMRecorder:    oomRecorder,
 			Provider:       historyDataSource,
 			PredictorMgr:   predictorMgr,
 			Recorder:       mgr.GetEventRecorderFor("recommendationrule-controller"),
@@ -403,6 +404,7 @@ func initControllers(oomRecorder oom.Recorder, mgr ctrl.Manager, opts *options.O
 			Client:         mgr.GetClient(),
 			RecommenderMgr: recommenderMgr,
 			ScaleClient:    scaleClient,
+			OOMRecorder:    oomRecorder,
 			Provider:       historyDataSource,
 			PredictorMgr:   predictorMgr,
 			Recorder:       mgr.GetEventRecorderFor("recommendation-trigger-controller"),

pkg/controller/recommendation/recommendation_rule_controller.go

@@ -32,6 +32,7 @@ import (
 	analysisv1alph1 "github.com/gocrane/api/analysis/v1alpha1"
 
 	"github.com/gocrane/crane/pkg/known"
+	"github.com/gocrane/crane/pkg/oom"
 	predictormgr "github.com/gocrane/crane/pkg/predictor"
 	"github.com/gocrane/crane/pkg/providers"
 	recommender "github.com/gocrane/crane/pkg/recommendation"
@@ -45,6 +46,7 @@ type RecommendationRuleController struct {
 	Recorder        record.EventRecorder
 	RestMapper      meta.RESTMapper
 	ScaleClient     scale.ScalesGetter
+	OOMRecorder     oom.Recorder
 	RecommenderMgr  recommender.RecommenderManager
 	PredictorMgr    predictormgr.Manager
 	kubeClient      kubernetes.Interface
@@ -199,7 +201,7 @@ func (c *RecommendationRuleController) doReconcile(ctx context.Context, recommen
 			}
 		}
 
-		go executeMission(ctx, &wg, c.RecommenderMgr, c.Provider, c.PredictorMgr, recommendationRule, identities, &currMissions[index], existingRecommendation, c.Client, c.ScaleClient, timeNow, newStatus.RunNumber)
+		go executeMission(ctx, &wg, c.RecommenderMgr, c.Provider, c.PredictorMgr, recommendationRule, identities, &currMissions[index], existingRecommendation, c.Client, c.ScaleClient, c.OOMRecorder, timeNow, newStatus.RunNumber)
 	}
 
 	wg.Wait()
@@ -410,7 +412,7 @@ func CreateRecommendationObject(recommendationRule *analysisv1alph1.Recommendati
 
 func executeMission(ctx context.Context, wg *sync.WaitGroup, recommenderMgr recommender.RecommenderManager, provider providers.History, predictorMgr predictormgr.Manager,
 	recommendationRule *analysisv1alph1.RecommendationRule, identities map[string]ObjectIdentity, mission *analysisv1alph1.RecommendationMission,
-	existingRecommendation *analysisv1alph1.Recommendation, client client.Client, scaleClient scale.ScalesGetter, timeNow metav1.Time, currentRunNumber int32) {
+	existingRecommendation *analysisv1alph1.Recommendation, client client.Client, scaleClient scale.ScalesGetter, oomRecorder oom.Recorder, timeNow metav1.Time, currentRunNumber int32) {
 	defer func() {
 		mission.LastStartTime = &timeNow
 		klog.Infof("Mission message: %s", mission.Message)
@@ -444,7 +446,7 @@ func executeMission(ctx context.Context, wg *sync.WaitGroup, recommenderMgr reco
 			Labels:     identities[k].Labels,
 			Object:     identities[k].Object,
 		}
-		recommendationContext := framework.NewRecommendationContext(ctx, identity, recommendationRule, predictorMgr, p, recommendation, client, scaleClient)
+		recommendationContext := framework.NewRecommendationContext(ctx, identity, recommendationRule, predictorMgr, p, recommendation, client, scaleClient, oomRecorder)
 		err = recommender.Run(&recommendationContext, r)
 		if err != nil {
 			mission.Message = fmt.Sprintf("Failed to run recommendation flow in recommender %s: %s", r.Name(), err.Error())

pkg/controller/recommendation/recommendation_trigger_controller.go

@@ -21,6 +21,7 @@ import (
 
 	analysisv1alpha1 "github.com/gocrane/api/analysis/v1alpha1"
 
+	"github.com/gocrane/crane/pkg/oom"
 	predictormgr "github.com/gocrane/crane/pkg/predictor"
 	"github.com/gocrane/crane/pkg/providers"
 	recommender "github.com/gocrane/crane/pkg/recommendation"
@@ -33,6 +34,7 @@ type RecommendationTriggerController struct {
 	Recorder        record.EventRecorder
 	RecommenderMgr  recommender.RecommenderManager
 	ScaleClient     scale.ScalesGetter
+	OOMRecorder     oom.Recorder
 	discoveryClient discovery.DiscoveryInterface
 	dynamicClient   dynamic.Interface
 	PredictorMgr    predictormgr.Manager
@@ -104,7 +106,7 @@ func (c *RecommendationTriggerController) Reconcile(ctx context.Context, req ctr
 		return ctrl.Result{}, nil
 	}
 
-	executeMission(context.TODO(), nil, c.RecommenderMgr, c.Provider, c.PredictorMgr, recommendationRule, identities, &newStatus.Recommendations[currentMissionIndex], recommendation, c.Client, c.ScaleClient, metav1.Now(), newStatus.RunNumber)
+	executeMission(context.TODO(), nil, c.RecommenderMgr, c.Provider, c.PredictorMgr, recommendationRule, identities, &newStatus.Recommendations[currentMissionIndex], recommendation, c.Client, c.ScaleClient, c.OOMRecorder, metav1.Now(), newStatus.RunNumber)
 	if newStatus.Recommendations[currentMissionIndex].Message != "Success" {
 		err = c.Client.Delete(context.TODO(), recommendation)
 		if err != nil {

pkg/recommendation/config/config.go

@@ -60,14 +60,17 @@ func loadConfigFromBytes(buf []byte) (*apis.RecommenderConfiguration, error) {
 	return config, nil
 }
 
-func GetRecommendersFromConfiguration(file string) ([]apis.Recommender, error) {
+func GetRecommendersFromConfiguration(file string) (map[string]apis.Recommender, error) {
 	config, err := LoadRecommenderConfigFromFile(file)
 	if err != nil {
 		klog.Errorf("load recommender configuration failed, %v", err)
 		return nil, err
 	}
-
-	return config.Recommenders, nil
+	recommenders := make(map[string]apis.Recommender, len(config.Recommenders))
+	for _, recommender := range config.Recommenders {
+		recommenders[recommender.Name] = recommender
+	}
+	return recommenders, nil
 }
 
 func GetKeysOfMap(m map[string]string) (keys []string) {

pkg/recommendation/framework/context.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/gocrane/crane/pkg/common"
 	"github.com/gocrane/crane/pkg/metricnaming"
+	"github.com/gocrane/crane/pkg/oom"
 	"github.com/gocrane/crane/pkg/prediction/config"
 	predictormgr "github.com/gocrane/crane/pkg/predictor"
 	"github.com/gocrane/crane/pkg/providers"
@@ -62,6 +63,8 @@ type RecommendationContext struct {
 	RestMapper meta.RESTMapper
 	// ScalesGetter
 	ScaleClient scale.ScalesGetter
+	// oom.Recorder
+	OOMRecorder oom.Recorder
 	// Scale
 	Scale *autoscalingapiv1.Scale
 	// Pods in recommendation
@@ -72,7 +75,7 @@ type RecommendationContext struct {
 	EHPA *autoscalingapi.EffectiveHorizontalPodAutoscaler
 }
 
-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 {
+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, oomRecorder oom.Recorder) RecommendationContext {
 	return RecommendationContext{
 		Context:            context,
 		Identity:           identity,
@@ -85,6 +88,8 @@ func NewRecommendationContext(context context.Context, identity ObjectIdentity,
 		Client:             client,
 		RestMapper:         client.RESTMapper(),
 		ScaleClient:        scaleClient,
+		OOMRecorder:        oomRecorder,
+		//CancelCh:       context.Done(),
 	}
 }
 

pkg/recommendation/manager.go

@@ -8,16 +8,14 @@ import (
 	"k8s.io/klog/v2"
 
 	analysisv1alph1 "github.com/gocrane/api/analysis/v1alpha1"
-	"github.com/gocrane/crane/pkg/oom"
-	"github.com/gocrane/crane/pkg/providers"
 	"github.com/gocrane/crane/pkg/recommendation/config"
 	"github.com/gocrane/crane/pkg/recommendation/framework"
 	"github.com/gocrane/crane/pkg/recommendation/recommender"
 	"github.com/gocrane/crane/pkg/recommendation/recommender/apis"
-	"github.com/gocrane/crane/pkg/recommendation/recommender/hpa"
-	"github.com/gocrane/crane/pkg/recommendation/recommender/idlenode"
-	"github.com/gocrane/crane/pkg/recommendation/recommender/replicas"
-	"github.com/gocrane/crane/pkg/recommendation/recommender/resource"
+	_ "github.com/gocrane/crane/pkg/recommendation/recommender/hpa"
+	_ "github.com/gocrane/crane/pkg/recommendation/recommender/idlenode"
+	_ "github.com/gocrane/crane/pkg/recommendation/recommender/replicas"
+	_ "github.com/gocrane/crane/pkg/recommendation/recommender/resource"
 )
 
 type RecommenderManager interface {
@@ -27,10 +25,9 @@ type RecommenderManager interface {
 	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 {
+func NewRecommenderManager(recommendationConfiguration string) RecommenderManager {
 	m := &manager{
 		recommendationConfiguration: recommendationConfiguration,
-		oomRecorder:                 oomRecorder,
 	}
 
 	m.loadConfigFile() // nolint:errcheck
@@ -51,8 +48,7 @@ type manager struct {
 	recommendationConfiguration string
 
 	lock               sync.Mutex
-	recommenderConfigs []apis.Recommender
-	oomRecorder        oom.Recorder
+	recommenderConfigs map[string]apis.Recommender
 }
 
 func (m *manager) GetRecommender(recommenderName string) (recommender.Recommender, error) {
@@ -63,22 +59,10 @@ func (m *manager) GetRecommenderWithRule(recommenderName string, recommendationR
 	m.lock.Lock()
 	defer m.lock.Unlock()
 
-	for _, r := range m.recommenderConfigs {
-		if r.Name == recommenderName {
-			switch recommenderName {
-			case recommender.ReplicasRecommender:
-				return replicas.NewReplicasRecommender(r, recommendationRule)
-			case recommender.HPARecommender:
-				return hpa.NewHPARecommender(r, recommendationRule)
-			case recommender.ResourceRecommender:
-				return resource.NewResourceRecommender(r, recommendationRule, m.oomRecorder)
-			case recommender.IdleNodeRecommender:
-				return idlenode.NewIdleNodeRecommender(r, recommendationRule)
-			default:
-				return nil, fmt.Errorf("unknown recommender name: %s", recommenderName)
-			}
-		}
+	if recommenderConfig, ok := m.recommenderConfigs[recommenderName]; ok {
+		return recommender.GetRecommenderProvider(recommenderName, recommenderConfig, recommendationRule)
 	}
+
 	return nil, fmt.Errorf("unknown recommender name: %s", recommenderName)
 }
 
@@ -133,12 +117,12 @@ func (m *manager) loadConfigFile() error {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 
-	apiRecommenders, err := config.GetRecommendersFromConfiguration(m.recommendationConfiguration)
+	recommenderConfigs, err := config.GetRecommendersFromConfiguration(m.recommendationConfiguration)
 	if err != nil {
 		klog.ErrorS(err, "Failed to load recommendation config file", "file", m.recommendationConfiguration)
 		return err
 	}
-	m.recommenderConfigs = apiRecommenders
+	m.recommenderConfigs = recommenderConfigs
 	klog.Info("Recommendation Config updated.")
 	return nil
 }

pkg/recommendation/recommender/hpa/registry.go

@@ -21,12 +21,16 @@ type HPARecommender struct {
 	MaxReplicasFactor       float64
 }
 
+func init() {
+	recommender.RegisterRecommenderProvider(recommender.HPARecommender, NewHPARecommender)
+}
+
 func (rr *HPARecommender) Name() string {
 	return recommender.HPARecommender
 }
 
 // NewHPARecommender create a new hpa recommender.
-func NewHPARecommender(recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (*HPARecommender, error) {
+func NewHPARecommender(recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (recommender.Recommender, error) {
 	recommender = config.MergeRecommenderConfigFromRule(recommender, recommendationRule)
 
 	predictableEnabled, err := recommender.GetConfigBool("predictable", false)
@@ -70,7 +74,7 @@ func NewHPARecommender(recommender apis.Recommender, recommendationRule analysis
 	}
 
 	return &HPARecommender{
-		*replicasRecommender,
+		*(replicasRecommender.(*replicas.ReplicasRecommender)),
 		predictableEnabled,
 		referenceHpaEnabled,
 		minCpuUsageThresholdFloat,

pkg/recommendation/recommender/idlenode/registry.go

@@ -25,12 +25,16 @@ type IdleNodeRecommender struct {
 	memoryUsageUtilization   float64
 }
 
+func init() {
+	recommender.RegisterRecommenderProvider(recommender.IdleNodeRecommender, NewIdleNodeRecommender)
+}
+
 func (inr *IdleNodeRecommender) Name() string {
 	return recommender.IdleNodeRecommender
 }
 
 // NewIdleNodeRecommender create a new idle node recommender.
-func NewIdleNodeRecommender(recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (*IdleNodeRecommender, error) {
+func NewIdleNodeRecommender(recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (recommender.Recommender, error) {
 	recommender = config.MergeRecommenderConfigFromRule(recommender, recommendationRule)
 
 	cpuRequestUtilization, err := recommender.GetConfigFloat(cpuRequestUtilizationKey, 0)

pkg/recommendation/recommender/recommenders.go

@@ -0,0 +1,44 @@
+package recommender
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/klog/v2"
+
+	analysisv1alph1 "github.com/gocrane/api/analysis/v1alpha1"
+	"github.com/gocrane/crane/pkg/recommendation/recommender/apis"
+)
+
+type Factory func(apis.Recommender, analysisv1alph1.RecommendationRule) (Recommender, error)
+
+// All registered Recommender providers.
+var (
+	providersMutex sync.Mutex
+	providers      = make(map[string]Factory)
+)
+
+// RegisterRecommenderProvider registers a recommender.Factory by name.  This
+// is expected to happen during app startup.
+func RegisterRecommenderProvider(name string, recommender Factory) {
+	providersMutex.Lock()
+	defer providersMutex.Unlock()
+	if _, found := providers[name]; found {
+		klog.Fatalf("recommender provider %q was registered twice", name)
+	}
+	klog.V(1).Infof("Registered recommender provider %q", name)
+	providers[name] = recommender
+}
+
+// GetRecommenderProvider creates an instance of the named Recommender provider, or nil if
+// the name is unknown.  The error return is only used if the named provider
+// was known but failed to initialize.
+func GetRecommenderProvider(recommenderName string, recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (Recommender, error) {
+	providersMutex.Lock()
+	defer providersMutex.Unlock()
+	f, found := providers[recommenderName]
+	if !found {
+		return nil, fmt.Errorf("unknown recommender name: %s", recommenderName)
+	}
+	return f(recommender, recommendationRule)
+}

pkg/recommendation/recommender/replicas/registry.go

@@ -22,12 +22,16 @@ type ReplicasRecommender struct {
 	MemTargetUtilization float64
 }
 
+func init() {
+	recommender.RegisterRecommenderProvider(recommender.ReplicasRecommender, NewReplicasRecommender)
+}
+
 func (rr *ReplicasRecommender) Name() string {
 	return recommender.ReplicasRecommender
 }
 
 // NewReplicasRecommender create a new replicas recommender.
-func NewReplicasRecommender(recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (*ReplicasRecommender, error) {
+func NewReplicasRecommender(recommender apis.Recommender, recommendationRule analysisv1alph1.RecommendationRule) (recommender.Recommender, error) {
 	recommender = config.MergeRecommenderConfigFromRule(recommender, recommendationRule)
 
 	workloadMinReplicasInt, err := recommender.GetConfigInt("workload-min-replicas", 1)

pkg/recommendation/recommender/resource/recommend.go

@@ -94,7 +94,7 @@ func (rr *ResourceRecommender) Recommend(ctx *framework.RecommendationContext) e
 	var newContainers []corev1.Container
 	var oldContainers []corev1.Container
 
-	oomRecords, err := rr.oomRecorder.GetOOMRecord()
+	oomRecords, err := ctx.OOMRecorder.GetOOMRecord()
 	if err != nil {
 		return err
 	}